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

refactor: 重构所有面板 update_from_config 为文件系统扫描模式

Browse Source 
=== 核心变更 ===
- _step_path_resolver 新增 scan_work_dir_for_input() 统一文件扫描工具
- 基于 _SCAN_TABLE 映射表，按 output_type 自动扫描 work_dir 子目录
- 支持扩展名匹配（.dat/.tif/.bsq/.csv 等）和文件名关键词匹配
- 按 mtime 排序，返回最新匹配文件

=== 各面板重构 ===
- step2/3/4/6: 废弃 main_window.stepX_panel 跨面板读取
  统一改为 scan_work_dir_for_input(work_dir, 'water_mask/glint_mask/deglint_image')
- step6: 移除对 step1/2/3/5 panel 的4处跨面板依赖
- step8: 替换 _resolve_training_csv_from_workdir 为 scan_work_dir_for_input
  优先级：training_spectra_indices -> training_spectra
- step9: 替换 _resolve_latest_wqi_test_csv 为 scan_work_dir_for_input
  废弃 factory.get_panel('step4_sampling') 和 get_panel('step8_ml_train')
- step10: 替换3层回退链为 scan_work_dir_for_input('sampling_points')
- step11: 替换 factory.get_panel('step9')/('step1') 为文件扫描

=== 删除的冗余方法 ===
- step8._resolve_training_csv_from_workdir (~55行)
- step9._resolve_latest_wqi_test_csv (~56行)

=== 数据流原则 ===
1. pipeline context 为第一顺位（执行时内存状态）
2. 文件系统扫描为回退（仅信任硬盘上真实存在的文件）
3. 彻底禁止面板间 UI 控件互相读取
This commit is contained in:
duxin
2026-06-30 09:53:23 +08:00
parent 1bdd623fe7
commit 73cb019a4a
9 changed files with 194 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
src/gui/panels/_step_path_resolver.py
View File

@ -183,9 +183,91 @@ def resolve_subdir(work_dir, subdir_key: str) -> str:
return wd
# ═══════════════════════════════════════════════════════════════
# 文件系统扫描表 —— 定义每个上游产出类型对应的搜索策略
# ═══════════════════════════════════════════════════════════════
_SCAN_TABLE = {
# output_type → (subdir_key, extensions, file_name_matcher)
'water_mask': ('water_mask', ['.dat', '.tif', '.tiff'], None),
'glint_mask': ('glint_detection', ['.dat'], 'severe_glint'),
'deglint_image': ('deglint', ['.bsq', '.dat', '.tif', '.tiff'], None),
'sampling_points': ('sampling', ['.csv'], 'sampling_spectra'),
'processed_data': ('data_cleaning', ['.csv'], 'processed_data'),
'training_spectra': ('spectral_feature',['.csv'], 'training_spectra'),
'training_spectra_indices': ('indices', ['.csv'], 'training_spectra_indices'),
'ml_models_dir': ('supervised_models', None, None), # 目录
'ml_predictions_dir': ('ml_prediction', None, None), # 目录
'water_index_csv_dir': ('watercolor', None, None), # 目录
'visualization_dir': ('visualization', None, None), # 目录
'reference_img': ('water_mask', ['.bsq', '.dat', '.tif', '.tiff'], None),
}
def scan_work_dir_for_input(work_dir: str, output_type: str):
"""基于文件系统扫描查找上游步骤的产出文件。
这是 update_from_config 重构的核心工具函数。
仅信任硬盘上真实存在的文件，彻底消除面板间 UI 控件互相读取。
Args:
work_dir: 工作目录路径
output_type: 产出类型，如 'water_mask', 'deglint_image', 'sampling_points'
Returns:
找到的文件/目录绝对路径字符串，未找到返回 None
"""
if not work_dir:
return None
wd = Path(work_dir)
entry = _SCAN_TABLE.get(output_type)
if entry is None:
return None
subdir_key, extensions, matcher = entry
target_dir = wd / _FALLBACK_DIR_TABLE.get(subdir_key, subdir_key)
if not target_dir.is_dir():
return None
# 目录类型（extensions 为 None）→ 只要目录存在就返回
if extensions is None:
return str(target_dir).replace('\\', '/')
# 扫描目录中的文件
candidates = []
for ext in extensions:
for f in target_dir.glob(f'*{ext}'):
if not f.is_file():
continue
candidates.append(f)
# 也搜 ext 的大写变体
for f in target_dir.glob(f'*{ext.upper()}'):
if not f.is_file():
continue
candidates.append(f)
if not candidates:
return None
# 按 matcher 优先级 + mtime 排序
if matcher:
matched = [f for f in candidates if matcher.lower() in f.stem.lower()]
if matched:
matched.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return str(matched[0]).replace('\\', '/')
# matcher 未命中时，仍返回最新文件（宽松匹配）
candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return str(candidates[0]).replace('\\', '/')
# 无 matcher → 返回最新的
candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return str(candidates[0]).replace('\\', '/')
__all__ = [
'STEP_DATA_SOURCE',
'resolve_step_widget',
'get_step_output_path',
'resolve_subdir',
'scan_work_dir_for_input',
]

42
src/gui/panels/step10_watercolor_panel.py
View File

@ -24,7 +24,7 @@ from typing import Dict, List, Optional
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from _step_path_resolver import resolve_subdir
from _step_path_resolver import resolve_subdir, scan_work_dir_for_input
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGridLayout, QFormLayout,
@ -554,9 +554,7 @@ class Step10WatercolorPanel(QWidget):
else:
self.work_dir = None
main_window = self.window()
# 1. 优先从 pipeline.step_outputs 取 Step 4 的采样点 CSV 路径
# 1. 采样点 CSV：优先 pipeline.step_outputs，回退文件系统扫描
sampling_path = None
if pipeline and hasattr(pipeline, 'step_outputs'):
step4_out = pipeline.step_outputs.get('step4_sampling', {})
@ -565,40 +563,12 @@ class Step10WatercolorPanel(QWidget):
or step4_out.get('output_path')
or step4_out.get('output_file')
)
# 2. 回退：直接读 step4_sampling panel 的 output_file widget
# 2026-06-30：panel widget 可能含幽灵占位路径，仅当文件确实存在时才采纳
if not sampling_path and main_window:
step4_widget = getattr(main_window, 'step4_sampling', None)
if step4_widget and hasattr(step4_widget, 'output_file'):
candidate = step4_widget.output_file.get_path()
if candidate and os.path.isfile(candidate):
sampling_path = candidate
if not sampling_path:
# 通过 _panel_factory 懒加载查找
factory = getattr(main_window, '_panel_factory', None)
if factory:
step4_panel = factory.get_panel('step4_sampling')
if step4_panel and hasattr(step4_panel, 'output_file'):
candidate = step4_panel.output_file.get_path()
if candidate and os.path.isfile(candidate):
sampling_path = candidate
# 3. 终极回退：扫描 work_dir/4_sampling/sampling_spectra.csv
if not sampling_path and self.work_dir:
candidate = resolve_subdir(self.work_dir, 'sampling_csv_path')
if os.path.isfile(candidate):
sampling_path = candidate
sampling_path = scan_work_dir_for_input(self.work_dir, 'sampling_points')
if sampling_path and os.path.exists(str(sampling_path)):
self.sampling_csv_file.set_path(str(sampling_path))
# 填入 UI
if sampling_path:
if not os.path.isabs(sampling_path):
sampling_path = os.path.join(
self.work_dir or '', sampling_path
).replace('\\', '/')
self.sampling_csv_file.set_path(sampling_path)
# 自动填入输出目录（仅在为空时填入默认路径，不创建目录）
# 2. 自动填入输出目录（仅在为空时填入默认路径，不创建目录）
if self.work_dir and not self.output_dir.get_path():
out_dir = os.path.join(
self.work_dir, '10_WaterIndex_CSV'

35
src/gui/panels/step11_map_panel.py
View File

@ -14,7 +14,7 @@ from typing import List, Optional
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from src.gui.panels._step_path_resolver import resolve_subdir, get_step_output_path
from src.gui.panels._step_path_resolver import resolve_subdir, get_step_output_path, scan_work_dir_for_input
from PyQt5.QtCore import Qt, QThread, pyqtSignal
from PyQt5.QtWidgets import (
@ -478,32 +478,19 @@ class Step11MapPanel(QWidget):
if work_dir:
self.work_dir = work_dir
main_window = self.window()
factory = getattr(main_window, '_panel_factory', None) if main_window else None
if not factory: return
# 1. 安全抓取 Step 9 的预测 CSV 目录（仅当目录确实存在）
step9_panel = factory.get_panel('step9_ml_predict')
if step9_panel and hasattr(step9_panel, 'output_file'):
path = step9_panel.output_file.get_path()
if path and os.path.isdir(path):
self.prediction_csv_dir_edit.setText(path)
# 1. 预测 CSV 目录：文件系统扫描
if self.work_dir:
pred_dir = scan_work_dir_for_input(self.work_dir, 'ml_predictions_dir')
if pred_dir and os.path.isdir(str(pred_dir)):
self.prediction_csv_dir_edit.setText(str(pred_dir))
self.batch_mode_combo.setCurrentIndex(1)
# 2. 安全抓取 Step 1 的真实掩膜文件（仅当文件确实存在）
step1_panel = factory.get_panel('step1')
if step1_panel:
use_ndwi = step1_panel.use_ndwi_radio.isChecked()
if use_ndwi and hasattr(step1_panel, 'output_file'):
path = step1_panel.output_file.get_path()
elif not use_ndwi and hasattr(step1_panel, 'mask_file'):
path = step1_panel.mask_file.get_path()
else:
path = ""
# 2. 边界文件（水体掩膜）：文件系统扫描
if self.work_dir:
boundary_path = scan_work_dir_for_input(self.work_dir, 'water_mask')
existing = self.boundary_file.get_path()
if path and not existing and os.path.exists(path):
self.boundary_file.set_path(path)
if boundary_path and not existing and os.path.exists(str(boundary_path)):
self.boundary_file.set_path(str(boundary_path))
# 3. 生成第 11 步的输出目录（仅在为空时填入默认路径，不创建目录）
if hasattr(self, 'work_dir') and self.work_dir and not self.output_dir.get_path():

24
src/gui/panels/step2_panel.py
View File

@ -11,7 +11,7 @@ from pathlib import Path
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from _step_path_resolver import resolve_subdir
from _step_path_resolver import resolve_subdir, scan_work_dir_for_input
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QFormLayout,
@ -178,24 +178,26 @@ class Step2Panel(QWidget):
else:
self.work_dir = None
# ── 水域掩膜输入 ──
# 优先：pipeline context（执行时内存中的确切状态）
mask_path = None
if pipeline and hasattr(pipeline, 'water_mask_path') and pipeline.water_mask_path:
mask_path = pipeline.water_mask_path
elif pipeline and hasattr(pipeline, 'step_outputs'):
step1_out = pipeline.step_outputs.get('step1', {})
mask_path = step1_out.get('water_mask') or step1_out.get('output_path')
main_window = self.window()
if not mask_path and hasattr(main_window, 'step1_panel'):
if main_window.step1_panel.use_ndwi_radio.isChecked():
mask_path = main_window.step1_panel.output_file.get_path()
else:
mask_path = main_window.step1_panel.mask_file.get_path()
# 回退：基于 work_dir 的文件系统扫描（仅信任硬盘上真实存在的文件）
if not mask_path or not os.path.exists(mask_path):
mask_path = scan_work_dir_for_input(self.work_dir, 'water_mask')
if mask_path:
# 填入 UI
if mask_path and os.path.exists(mask_path):
if not os.path.isabs(mask_path):
mask_path = os.path.join(self.work_dir or '', mask_path).replace('\\', '/')
# 仅当上游文件确实存在时才自动填入（防止幽灵路径级联扩散）
if os.path.exists(mask_path):
self.water_mask_file.set_path(mask_path)
self.water_mask_file.set_path(mask_path)
# ── 输出路径 ──
if self.work_dir:
if not self.output_file.get_path():
output_dir = resolve_subdir(self.work_dir, 'glint_detection')

29
src/gui/panels/step3_panel.py
View File

@ -11,7 +11,7 @@ from pathlib import Path
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from _step_path_resolver import resolve_subdir
from _step_path_resolver import resolve_subdir, scan_work_dir_for_input
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QFormLayout,
@ -307,20 +307,23 @@ class Step3Panel(QWidget):
else:
self.work_dir = None
main_window = self.window()
if hasattr(main_window, 'step1_panel'):
if main_window.step1_panel.use_ndwi_radio.isChecked():
mask_path = main_window.step1_panel.output_file.get_path()
else:
mask_path = main_window.step1_panel.mask_file.get_path()
# ── 水域掩膜输入 ──
# 优先：pipeline context
mask_path = None
if pipeline and hasattr(pipeline, 'step_outputs'):
step1_out = pipeline.step_outputs.get('step1', {})
mask_path = step1_out.get('water_mask') or step1_out.get('output_path')
if mask_path:
if not os.path.isabs(mask_path):
mask_path = os.path.join(self.work_dir or '', mask_path).replace('\\', '/')
# 仅当上游文件确实存在时才自动填入（防止幽灵路径级联扩散）
if os.path.exists(mask_path):
self.water_mask_file.set_path(mask_path)
# 回退：文件系统扫描 1_water_mask/
if not mask_path or not os.path.exists(mask_path):
mask_path = scan_work_dir_for_input(self.work_dir, 'water_mask')
if mask_path and os.path.exists(mask_path):
if not os.path.isabs(mask_path):
mask_path = os.path.join(self.work_dir or '', mask_path).replace('\\', '/')
self.water_mask_file.set_path(mask_path)
# ── 输出路径 ──
if self.work_dir:
if not self.output_file.get_path():
output_dir = resolve_subdir(self.work_dir, 'deglint')

58
src/gui/panels/step4_sampling_panel.py
View File

@ -25,7 +25,7 @@ plt.rcParams['axes.unicode_minus'] = False
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from _step_path_resolver import resolve_subdir
from _step_path_resolver import resolve_subdir, scan_work_dir_for_input
from PyQt5.QtCore import QTimer, Qt
from PyQt5.QtWidgets import (
@ -288,57 +288,45 @@ class Step4SamplingPanel(QWidget):
else:
self.work_dir = None
main_window = self.window()
# ── 去耀斑影像输入 ──
# 优先：pipeline context
deglint_path = None
if pipeline and hasattr(pipeline, 'step_outputs'):
step3_outputs = getattr(pipeline, 'step_outputs', {}).get('step3', {})
step3_out = pipeline.step_outputs.get('step3', {})
deglint_path = (
step3_outputs.get('deglint_image') or step3_outputs.get('output_path') or
step3_outputs.get('output_file') or step3_outputs.get('deglint_img_path')
step3_out.get('deglint_image') or step3_out.get('output_path') or
step3_out.get('output_file') or step3_out.get('deglint_img_path')
)
if not deglint_path and hasattr(main_window, 'step3_panel'):
deglint_path = main_window.step3_panel.output_file.get_path()
if deglint_path:
# 回退：文件系统扫描 3_deglint/
if not deglint_path or not os.path.exists(deglint_path):
deglint_path = scan_work_dir_for_input(self.work_dir, 'deglint_image')
if deglint_path and os.path.exists(deglint_path):
if not os.path.isabs(deglint_path):
deglint_path = os.path.join(self.work_dir or '', deglint_path).replace('\\', '/')
if os.path.exists(deglint_path):
self.deglint_img_file.set_path(deglint_path)
self.deglint_img_file.set_path(deglint_path)
# ── 水域掩膜输入 ──
# 优先：pipeline context
water_mask_path = None
if pipeline and hasattr(pipeline, 'step_outputs'):
step1_outputs = getattr(pipeline, 'step_outputs', {}).get('step1', {})
step1_out = pipeline.step_outputs.get('step1', {})
water_mask_path = (
step1_outputs.get('water_mask') or step1_outputs.get('output_path') or step1_outputs.get(
'output_file')
step1_out.get('water_mask') or step1_out.get('output_path') or
step1_out.get('output_file')
)
if not water_mask_path and hasattr(main_window, 'step1_panel'):
water_mask_path = main_window.step1_panel.output_file.get_path()
if not water_mask_path and self.work_dir:
mask_dir = resolve_subdir(self.work_dir, 'water_mask')
if os.path.isdir(mask_dir):
dat_files = [f for f in os.listdir(mask_dir) if f.lower().endswith('.dat')]
if dat_files:
water_mask_path = os.path.join(mask_dir, dat_files[0]).replace('\\', '/')
# 回退：文件系统扫描 1_water_mask/
if not water_mask_path or not os.path.exists(water_mask_path):
water_mask_path = scan_work_dir_for_input(self.work_dir, 'water_mask')
if not water_mask_path and self.work_dir:
input_test_dir = os.path.join(self.work_dir, "input-test")
if os.path.isdir(input_test_dir):
dat_files = [f for f in os.listdir(input_test_dir) if f.lower().endswith('.dat')]
for f in dat_files:
if 'water_mask_from_shp' in f.lower():
water_mask_path = os.path.join(input_test_dir, f).replace('\\', '/')
break
if not water_mask_path and dat_files:
water_mask_path = os.path.join(input_test_dir, dat_files[0]).replace('\\', '/')
if water_mask_path:
if water_mask_path and os.path.exists(water_mask_path):
if not os.path.isabs(water_mask_path):
water_mask_path = os.path.join(self.work_dir or '', water_mask_path).replace('\\', '/')
if os.path.exists(water_mask_path):
self.water_mask_file.set_path(water_mask_path)
self.water_mask_file.set_path(water_mask_path)
# ── 输出路径 ──
if self.work_dir and not self.output_file.get_path():
output_path = resolve_subdir(self.work_dir, 'sampling_csv_path')
self.output_file.set_path(output_path.replace('\\', '/'))

80
src/gui/panels/step6_feature_panel.py
View File

@ -12,7 +12,7 @@ from pathlib import Path
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from src.gui.panels._step_path_resolver import resolve_subdir
from src.gui.panels._step_path_resolver import resolve_subdir, scan_work_dir_for_input
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QFormLayout,
@ -218,33 +218,22 @@ class Step6FeaturePanel(QWidget):
else:
self.work_dir = None
# 1. 水体掩膜：优先 pipeline，回退文件系统扫描
mask_path = None
if pipeline and hasattr(pipeline, 'water_mask_path') and pipeline.water_mask_path:
mask_path = pipeline.water_mask_path
if not mask_path and self.work_dir:
mask_path = scan_work_dir_for_input(self.work_dir, 'water_mask')
if mask_path and os.path.exists(str(mask_path)):
self.water_mask_file.set_path(str(mask_path))
main_window = self.window()
if not mask_path and hasattr(main_window, 'step1_panel'):
if main_window.step1_panel.use_ndwi_radio.isChecked():
mask_path = main_window.step1_panel.output_file.get_path()
else:
mask_path = main_window.step1_panel.mask_file.get_path()
if mask_path and not os.path.isabs(mask_path):
mask_path = os.path.join(self.work_dir or '', mask_path).replace('\\', '/')
if mask_path:
if not os.path.isabs(mask_path):
mask_path = os.path.join(self.work_dir or '', mask_path).replace('\\', '/')
if os.path.exists(mask_path):
self.water_mask_file.set_path(mask_path)
if hasattr(main_window, 'step2_panel'):
glint_path = main_window.step2_panel.output_file.get_path()
if glint_path:
if not os.path.isabs(glint_path):
glint_path = os.path.join(self.work_dir or '', glint_path).replace('\\', '/')
if os.path.exists(glint_path):
self.glint_mask_file.set_path(glint_path)
# 2. 耀斑掩膜：文件系统扫描
if self.work_dir:
glint_path = scan_work_dir_for_input(self.work_dir, 'glint_mask')
if glint_path and os.path.exists(str(glint_path)):
self.glint_mask_file.set_path(str(glint_path))
# 3. 去耀斑影像：优先 pipeline.step_outputs，回退文件系统扫描
deglint_path = None
if pipeline and hasattr(pipeline, 'step_outputs'):
step3_outputs = getattr(pipeline, 'step_outputs', {}).get('step3', {})
@ -254,30 +243,22 @@ class Step6FeaturePanel(QWidget):
or step3_outputs.get('output_file')
or step3_outputs.get('deglint_img_path')
)
if not deglint_path and hasattr(main_window, 'step3_panel'):
step3_widget = getattr(main_window.step3_panel, 'output_file', None)
if step3_widget is not None and hasattr(step3_widget, 'get_path'):
deglint_path = step3_widget.get_path() or ""
if not deglint_path and self.work_dir:
deglint_dir = resolve_subdir(self.work_dir, 'deglint')
if os.path.isdir(deglint_dir):
bsq_files = [
f for f in os.listdir(deglint_dir)
if f.lower().endswith('.bsq')
]
bsq_files.sort(key=lambda n: (0 if 'goodman' in n.lower() else 1, n))
if bsq_files:
deglint_path = os.path.join(deglint_dir, bsq_files[0]).replace('\\', '/')
if deglint_path:
if not os.path.isabs(deglint_path):
deglint_path = os.path.join(self.work_dir or '', deglint_path).replace('\\', '/')
deglint_path = scan_work_dir_for_input(self.work_dir, 'deglint_image')
if deglint_path and os.path.exists(str(deglint_path)):
existing_deglint = self.deglint_img_file.get_path()
if (not existing_deglint or not existing_deglint.strip()) and os.path.exists(deglint_path):
self.deglint_img_file.set_path(deglint_path)
if (not existing_deglint or not existing_deglint.strip()):
self.deglint_img_file.set_path(str(deglint_path))
# 4. 处理后 CSV：文件系统扫描
if self.work_dir:
csv_path = scan_work_dir_for_input(self.work_dir, 'processed_data')
if csv_path and os.path.exists(str(csv_path)):
existing_csv = self.csv_file.get_path()
if (not existing_csv or not existing_csv.strip()):
self.csv_file.set_path(str(csv_path))
# 5. 输出路径
if self.work_dir and not self.output_file.get_path():
output_dir = resolve_subdir(self.work_dir, 'spectral_feature')
default_output_path = os.path.join(output_dir, "training_spectra.csv").replace('\\', '/')
@ -285,17 +266,6 @@ class Step6FeaturePanel(QWidget):
elif not self.work_dir:
self.output_file.set_path("")
if main_window and hasattr(main_window, 'step5_clean_panel'):
step5_clean_output_path = main_window.step5_clean_panel.output_file.get_path()
if step5_clean_output_path:
if not os.path.isabs(step5_clean_output_path):
step5_clean_output_path = os.path.join(
self.work_dir or '', step5_clean_output_path
).replace('\\', '/')
existing_csv = self.csv_file.get_path()
if (not existing_csv or not existing_csv.strip()) and os.path.exists(step5_clean_output_path):
self.csv_file.set_path(step5_clean_output_path)
def _on_run_single_clicked(self):
from src.gui.core.event_bus import global_event_bus

65
src/gui/panels/step8_ml_train_panel.py
View File

@ -12,7 +12,7 @@ from pathlib import Path
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from src.gui.panels._step_path_resolver import get_step_output_path, resolve_step_widget, resolve_subdir
from src.gui.panels._step_path_resolver import get_step_output_path, resolve_step_widget, resolve_subdir, scan_work_dir_for_input
import pandas as pd
@ -404,56 +404,6 @@ class Step8MlTrainPanel(QWidget):
continue
return None
def _resolve_training_csv_from_workdir(self):
"""根据工作目录智能挑选训练 CSV 路径。
优先级（从高到低）：
1) 7_Water_Quality_Indices/training_spectra_indices.csv（Step 7 WQI 增强版）
2) 10_WaterIndex_CSV/*training* / *training*indices*.csv（用户自定义带指数汇总）
3) 6_Spectral_Feature_Extraction/training_spectra.csv（Step 6 原始特征）
4) 7_Water_Quality_Indices/ 下任意 *training*.csv
"""
work_dir = self._get_default_work_dir()
if not work_dir:
return ""
from pathlib import Path
wd = Path(work_dir)
# 1) Step 7 输出的训练 WQI 增强版
step7_csv = wd / "7_Water_Quality_Indices" / "training_spectra_indices.csv"
if step7_csv.is_file():
return str(step7_csv).replace('\\', '/')
# 2) 10_WaterIndex_CSV 下任何带 "training" 关键词的 csv（用户在 Step 10 跑过训练集）
idx_dir = wd / "10_WaterIndex_CSV"
if idx_dir.is_dir():
candidates = sorted(
idx_dir.glob("*training*.csv"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
if candidates:
return str(candidates[0]).replace('\\', '/')
# 3) Step 6 原始光谱特征
step6_csv = wd / "6_Spectral_Feature_Extraction" / "training_spectra.csv"
if step6_csv.is_file():
return str(step6_csv).replace('\\', '/')
# 4) Step 7 目录下任何 training*.csv（兜底）
step7_dir = wd / "7_Water_Quality_Indices"
if step7_dir.is_dir():
candidates = sorted(
step7_dir.glob("*training*.csv"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
if candidates:
return str(candidates[0]).replace('\\', '/')
return ""
def browse_output_path(self):
"""浏览输出模型目录"""
work_dir = getattr(self, 'work_dir', "")
@ -538,15 +488,14 @@ class Step8MlTrainPanel(QWidget):
else:
self.work_dir = None
# 1. 智能挑选训练 CSV（不再"强制"读 Step 6，而是优先 WQI 增强版）
# 优先级：Step 7 WQI > 10_WaterIndex_CSV/*training* > Step 6 原始光谱 > Step 7 兜底
# 修复目标：用户跑过 Step 7 后再回到 Step 8，UI 默认应指向带指数的训练集，
# 否则训练好的模型有 95 维（50 波段 + 45 WQI），下次回放变成只 50 维训练，特征维数错位。
# 1. 智能挑选训练 CSV：优先 WQI 增强版，回退原始光谱特征
existing_training_csv = self.training_csv_file.get_path()
if not existing_training_csv or not existing_training_csv.strip():
candidate = self._resolve_training_csv_from_workdir()
if candidate:
self.training_csv_file.set_path(candidate)
candidate = scan_work_dir_for_input(self.work_dir, 'training_spectra_indices')
if not candidate:
candidate = scan_work_dir_for_input(self.work_dir, 'training_spectra')
if candidate and os.path.exists(str(candidate)):
self.training_csv_file.set_path(str(candidate))
# 2. 自动填充输出目录（仅在为空时填入默认路径，不创建目录）
if self.work_dir and not self.output_path.get_path():

91
src/gui/panels/step9_ml_predict_panel.py
View File

@ -14,7 +14,7 @@ import pandas as pd
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
sys.path.insert(0, _HERE)
from _step_path_resolver import get_step_output_path, resolve_step_widget, resolve_subdir
from _step_path_resolver import get_step_output_path, resolve_step_widget, resolve_subdir, scan_work_dir_for_input
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QGroupBox, QFormLayout,
@ -337,89 +337,20 @@ class Step9MlPredictPanel(QWidget):
result[name] = self.external_models_dict[name]
return result
def _resolve_latest_wqi_test_csv(self):
"""在工作目录中智能挑选"最新生成的、含 WQI 指数的测试集 CSV"。
返回：找到则返回文件路径字符串；找不到返回 ""。
搜索策略（按优先级递减，命中即返回）：
1) 10_WaterIndex_CSV/*.csv — Step 10 输出目录（用户在 Step 10 跑过的产品）
2) 7_Water_Quality_Indices/*sampling*.csv / *test*.csv — 用户手动对采样点算过 WQI
3) work_dir 下任何 *indices*.csv / *wqi*.csv（不区分大小写）
4) work_dir 下任何 > 60 列的 csv（启发式：50 波段 + > 10 WQI 指数列）
5) 兜底空串，调用方回退到 Step 4 sampling_spectra.csv
多个候选时按 mtime 倒序选"最新生成的"。
"""
work_dir = self._get_default_work_dir()
if not work_dir:
return ""
wd = Path(work_dir)
found = []
# 1) 10_WaterIndex_CSV 下所有 csv（Step 10 输出）
idx_dir = wd / "10_WaterIndex_CSV"
if idx_dir.is_dir():
found.extend(idx_dir.glob("*.csv"))
# 2) 7_Water_Quality_Indices 下与采样/测试相关的 csv
qa_dir = wd / "7_Water_Quality_Indices"
if qa_dir.is_dir():
for pattern in ("*sampling*.csv", "*test*.csv", "*predict*.csv"):
found.extend(qa_dir.glob(pattern))
# 3) work_dir 直接子树下含 indices/wqi 关键词的 csv
for keyword in ("*indices*.csv", "*wqi*.csv", "*WQI*.csv"):
found.extend(wd.rglob(keyword))
# 4) 启发式：> 60 列的 csv（50 波段 + 至少 10 个指数）
try:
for csv in wd.rglob("*.csv"):
if csv in found:
continue
try:
head = pd.read_csv(csv, nrows=0)
if head.shape[1] > 60:
found.append(csv)
except Exception:
pass # 读取失败就跳过，不影响其它候选
except Exception:
pass
if not found:
return ""
# 去重 + 按 mtime 倒序排
uniq = {p.resolve(): p for p in found}.values()
sorted_paths = sorted(uniq, key=lambda p: p.stat().st_mtime, reverse=True)
return str(sorted_paths[0]).replace('\\', '/')
def update_from_config(self, work_dir=None, pipeline=None):
if work_dir: self.work_dir = work_dir
main_window = self.window()
factory = getattr(main_window, '_panel_factory', None) if main_window else None
# 1. 采样 CSV：文件系统扫描（sampling_points 即 sampling_spectra.csv）
if self.work_dir:
sampling_path = scan_work_dir_for_input(self.work_dir, 'sampling_points')
if sampling_path and os.path.exists(str(sampling_path)):
self.sampling_csv_file.set_path(str(sampling_path))
# 1. 智能挑选采样 CSV：优先"含 WQI 指数的测试集"（防止特征维度与训练时不匹配）
wqi_test_csv = self._resolve_latest_wqi_test_csv()
if wqi_test_csv:
self.sampling_csv_file.set_path(wqi_test_csv)
elif factory:
# 兜底：拿第 4 步的纯原始采样光谱（仅当文件确实存在）
step4_panel = factory.get_panel('step4_sampling')
if step4_panel and hasattr(step4_panel, 'output_file'):
path = step4_panel.output_file.get_path()
if path and os.path.exists(path):
self.sampling_csv_file.set_path(path)
# 2. 拿第 8 步的模型目录（仅当目录确实存在）
if factory:
step8_panel = factory.get_panel('step8_ml_train')
if step8_panel and hasattr(step8_panel, 'output_path'):
path = step8_panel.output_path.get_path()
if path and os.path.isdir(path):
self.models_dir_file.set_path(path)
# 2. 模型目录：文件系统扫描
if self.work_dir:
models_dir = scan_work_dir_for_input(self.work_dir, 'ml_models_dir')
if models_dir and os.path.isdir(str(models_dir)):
self.models_dir_file.set_path(str(models_dir))
# 3. 生成第 9 步的输出目录（仅在为空时填入默认路径，不创建目录）
if hasattr(self, 'work_dir') and self.work_dir and not self.output_file.get_path():