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

fix(gui): step8 QBrush崩溃修复 + step9 自动探测 Traditional_Indices 目录回填

Browse Source
This commit is contained in:
DXC
2026-06-09 13:13:01 +08:00
parent bf2496badc
commit 593719e7d0
2 changed files with 265 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
src/gui/panels/step8_panel.py
View File

@ -1,51 +1,53 @@
import os
import sys
import pandas as pd
import numpy as np
from pathlib import Path
from typing import Dict, List, Union
from typing import Dict, List, Optional, Tuple
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QGroupBox, QGridLayout,
QHBoxLayout, QLabel, QCheckBox, QPushButton, QMessageBox, QScrollArea
QHBoxLayout, QLabel, QCheckBox, QPushButton, QMessageBox,
QScrollArea, QListWidget, QListWidgetItem, QAbstractItemView,
QRadioButton, QButtonGroup
)
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QColor, QBrush, QFont
from src.gui.components.custom_widgets import FileSelectWidget
from src.gui.styles import ModernStylesheet
def get_resource_path(relative_path: str) -> str:
"""适配开发与 PyInstaller 环境的路径获取逻辑。
支持两种打包模式：
1. --onedir 模式：文件在 exe_root/_internal/ 下 → 检查 _internal 目录
2. --onefile 模式：文件在 sys._MEIPASS 平铺目录
"""
# 优先检查 PyInstaller onefile 模式（文件平铺在 _MEIPASS 下）
"""适配开发与 PyInstaller 环境的路径获取逻辑。"""
if hasattr(sys, '_MEIPASS'):
internal_path = os.path.join(sys._MEIPASS, '_internal', relative_path)
if os.path.exists(internal_path):
return internal_path
internal = os.path.join(sys._MEIPASS, '_internal', relative_path)
if os.path.exists(internal):
return internal
return os.path.join(sys._MEIPASS, relative_path)
# 兼容 PyInstaller onedir 模式的 _internal 目录（exe 同级目录下）
exe_dir = os.path.dirname(sys.executable)
internal_path = os.path.join(exe_dir, '_internal', relative_path)
if os.path.exists(internal_path):
return internal_path
internal = os.path.join(exe_dir, '_internal', relative_path)
if os.path.exists(internal):
return internal
# 开发环境下：基于当前文件 (step8_panel.py) 的绝对路径进行回溯
# 当前在 src/gui/panels/，目标在 src/gui/model/
base_dir = Path(__file__).resolve().parent.parent / "model"
target_path = base_dir / os.path.basename(relative_path)
return str(target_path)
return str(base_dir / os.path.basename(relative_path))
class Step8Panel(QWidget):
COLOR_RATIO = QColor(255, 255, 255)
COLOR_CONCENTRATION = QColor(220, 240, 255)
COLOR_HEADER = QColor(245, 245, 245)
def __init__(self, parent=None):
super().__init__(parent)
self.index_checkboxes: Dict[str, QCheckBox] = {}
# 标识为 waterindex.csv，目录跳转逻辑在 get_resource_path 中
self.work_dir: Optional[str] = None
self.builtin_formula_path = get_resource_path("waterindex.csv")
self._formula_type_map: Dict[str, str] = {}
self.init_ui()
# 延迟一小会儿加载，确保UI框架已就绪
self._auto_load_formulas()
def init_ui(self):
@ -53,13 +55,12 @@ class Step8Panel(QWidget):
main_layout.setContentsMargins(20, 20, 20, 20)
main_layout.setSpacing(10)
# 1. 路径展示区 (半透明只读)
# 1. 公式配置源 (只读)
path_group = QGroupBox("公式配置源 (内置)")
path_layout = QVBoxLayout()
self.formula_csv_widget = FileSelectWidget("内置CSV路径:", "CSV Files (*.csv)")
self.formula_csv_widget.set_path(self.builtin_formula_path)
self.formula_csv_widget.set_read_only(True)
# 视觉微调：提示用户这是内置的
self.formula_csv_widget.line_edit.setStyleSheet("background-color: #f0f0f0; color: #666;")
path_layout.addWidget(self.formula_csv_widget)
path_group.setLayout(path_layout)
@ -73,50 +74,78 @@ class Step8Panel(QWidget):
input_group.setLayout(input_layout)
main_layout.addWidget(input_group)
# 3. 公式选择区
# 3. 公式选择区 (分组 ListWidget)
self.formula_group = QGroupBox("待计算水质指数勾选")
formula_outer_layout = QVBoxLayout()
btn_layout = QHBoxLayout()
self.select_all_btn = QPushButton("全选")
self.deselect_all_btn = QPushButton("清空")
self.select_ratio_btn = QPushButton("仅选比值型")
self.select_conc_btn = QPushButton("仅选浓度型")
self.select_all_btn.clicked.connect(self.select_all_formulas)
self.deselect_all_btn.clicked.connect(self.deselect_all_formulas)
self.select_ratio_btn.clicked.connect(self._select_ratio_only)
self.select_conc_btn.clicked.connect(self._select_conc_only)
btn_layout.addWidget(self.select_all_btn)
btn_layout.addWidget(self.deselect_all_btn)
btn_layout.addWidget(self.select_ratio_btn)
btn_layout.addWidget(self.select_conc_btn)
btn_layout.addStretch()
self.refresh_button = QPushButton("手动重新加载公式")
self.refresh_button = QPushButton("重新加载")
self.refresh_button.clicked.connect(lambda: self.refresh_formulas(silent=False))
btn_layout.addWidget(self.refresh_button)
formula_outer_layout.addLayout(btn_layout)
# 核心滚动区
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setMinimumHeight(300) # 强制最小高度，防止塌陷
scroll.setMinimumHeight(280)
self.scroll_content = QWidget()
self.formula_layout = QGridLayout(self.scroll_content)
self.formula_layout.setAlignment(Qt.AlignTop) # 靠顶对齐
self.formula_layout = QVBoxLayout(self.scroll_content)
self.formula_layout.setContentsMargins(4, 4, 4, 4)
self.formula_layout.setSpacing(2)
self.formula_layout.setAlignment(Qt.AlignTop)
self.formula_list = QListWidget()
self.formula_list.setSelectionMode(QAbstractItemView.MultiSelection)
self.formula_list.itemChanged.connect(self._on_item_changed)
self.formula_layout.addWidget(self.formula_list)
scroll.setWidget(self.scroll_content)
formula_outer_layout.addWidget(scroll)
self.formula_group.setLayout(formula_outer_layout)
main_layout.addWidget(self.formula_group)
# 4. 输出与运行
output_group = QGroupBox("结果输出")
# 4. 输出选项
output_group = QGroupBox("输出模式")
output_layout = QVBoxLayout()
self.output_file_widget = FileSelectWidget("保存路径:", "CSV Files (*.csv)", mode="save")
output_layout.addWidget(self.output_file_widget)
output_group.setLayout(output_layout)
main_layout.addWidget(output_group)
mode_layout = QHBoxLayout()
self.mode_group = QButtonGroup()
self.radio_both = QRadioButton("两者皆出")
self.radio_wide = QRadioButton("仅宽表")
self.radio_single = QRadioButton("仅单文件")
self.mode_group.addButton(self.radio_both, 0)
self.mode_group.addButton(self.radio_wide, 1)
self.mode_group.addButton(self.radio_single, 2)
self.radio_both.setChecked(True)
mode_layout.addWidget(self.radio_both)
mode_layout.addWidget(self.radio_wide)
mode_layout.addWidget(self.radio_single)
mode_layout.addStretch()
output_layout.addLayout(mode_layout)
self.enable_checkbox = QCheckBox("启用计算流程")
self.enable_checkbox.setChecked(True)
main_layout.addWidget(self.enable_checkbox)
output_layout.addWidget(self.enable_checkbox)
output_group.setLayout(output_layout)
main_layout.addWidget(output_group)
# 5. 运行按钮
self.run_button = QPushButton("立即执行计算")
self.run_button.setStyleSheet(ModernStylesheet.get_button_stylesheet('success'))
self.run_button.setMinimumHeight(40)
@ -125,8 +154,17 @@ class Step8Panel(QWidget):
self.setLayout(main_layout)
def _on_item_changed(self, item: QListWidgetItem):
if item.checkState() == Qt.Checked:
color_data = item.data(Qt.UserRole + 1)
if isinstance(color_data, QColor):
item.setBackground(QBrush(QColor(color_data)))
else:
item.setBackground(QBrush(self.COLOR_RATIO))
else:
item.setBackground(QBrush(self.COLOR_RATIO))
def _auto_load_formulas(self):
"""启动时自动加载逻辑"""
if os.path.exists(self.builtin_formula_path):
self.refresh_formulas(silent=True)
else:
@ -135,91 +173,211 @@ class Step8Panel(QWidget):
def refresh_formulas(self, silent=False):
path = self.builtin_formula_path
if not os.path.exists(path):
if not silent: QMessageBox.warning(self, "错误", f"找不到内置公式文件:\n{path}")
if not silent:
QMessageBox.warning(self, "错误", f"找不到内置公式文件:\n{path}")
return
try:
# 清理旧列表
for i in reversed(range(self.formula_layout.count())):
widget = self.formula_layout.itemAt(i).widget()
if widget: widget.deleteLater()
self.index_checkboxes.clear()
# 鲁棒性读取：尝试不同编码
for encoding in ['utf-8', 'gbk', 'utf-8-sig']:
df = None
for enc in ('utf-8', 'gbk', 'utf-8-sig'):
try:
df = pd.read_csv(path, encoding=encoding)
if 'Formula_Name' in df.columns: break
except: continue
df = pd.read_csv(path, encoding=enc)
if 'Formula_Name' in df.columns:
break
except Exception:
continue
if 'Formula_Name' not in df.columns:
if not silent: QMessageBox.critical(self, "错误", "CSV文件缺少 'Formula_Name' 列")
if df is None or 'Formula_Name' not in df.columns:
if not silent:
QMessageBox.critical(self, "错误", "CSV缺少 'Formula_Name' 列")
return
names = df['Formula_Name'].dropna().unique().tolist()
self._formula_type_map.clear()
for _, row in df.iterrows():
name = str(row['Formula_Name']).strip()
if not name:
continue
ftype = str(row.get('Formula_Type', 'ratio')).strip().lower()
self._formula_type_map[name] = ftype
row, col = 0, 0
for name in names:
name = str(name).strip()
if not name: continue
cb = QCheckBox(name)
cb.setChecked(True)
self.index_checkboxes[name] = cb
self.formula_layout.addWidget(cb, row, col)
col += 1
if col >= 3:
col = 0
row += 1
self.formula_list.clear()
self.index_checkboxes.clear()
# 强制UI更新
self.scroll_content.adjustSize()
print(f"✅ 成功加载 {len(self.index_checkboxes)} 个公式")
for name, ftype in self._formula_type_map.items():
item = QListWidgetItem(name, self.formula_list)
item.setCheckState(Qt.Checked)
if ftype == 'concentration':
bg_color = QColor(220, 240, 255)
item.setData(Qt.UserRole + 1, bg_color)
item.setBackground(QBrush(bg_color))
else:
bg_color = self.COLOR_RATIO
item.setData(Qt.UserRole + 1, bg_color)
item.setBackground(QBrush(bg_color))
self.index_checkboxes[name] = item
self.formula_list.adjustSize()
print(f"✅ 加载 {len(self.index_checkboxes)} 个公式")
except Exception as e:
if not silent: QMessageBox.critical(self, "加载失败", f"原因: {str(e)}")
if not silent:
QMessageBox.critical(self, "加载失败", f"原因: {str(e)}")
def _select_ratio_only(self):
for name, item in self.index_checkboxes.items():
ftype = self._formula_type_map.get(name, 'ratio')
item.setCheckState(Qt.Checked if ftype == 'ratio' else Qt.Unchecked)
def _select_conc_only(self):
for name, item in self.index_checkboxes.items():
ftype = self._formula_type_map.get(name, 'ratio')
item.setCheckState(Qt.Checked if ftype == 'concentration' else Qt.Unchecked)
def select_all_formulas(self):
for cb in self.index_checkboxes.values(): cb.setChecked(True)
for item in self.index_checkboxes.values():
item.setCheckState(Qt.Checked)
def deselect_all_formulas(self):
for cb in self.index_checkboxes.values(): cb.setChecked(False)
for item in self.index_checkboxes.values():
item.setCheckState(Qt.Unchecked)
def get_config(self):
selected = [n for n, cb in self.index_checkboxes.items() if cb.isChecked()]
def get_config(self) -> Dict:
selected = [
name for name, item in self.index_checkboxes.items()
if item.checkState() == Qt.Checked
]
return {
'training_csv_path': self.training_data_widget.get_path(),
'formula_csv_file': self.builtin_formula_path,
'formula_names': selected,
'output_file': self.output_file_widget.get_path(),
'enabled': self.enable_checkbox.isChecked()
'enabled': self.enable_checkbox.isChecked(),
'output_mode': self.mode_group.checkedId(),
}
def set_config(self, config):
if 'training_csv_path' in config: self.training_data_widget.set_path(config['training_csv_path'])
def set_config(self, config: Dict):
if 'training_csv_path' in config:
self.training_data_widget.set_path(config['training_csv_path'])
if 'formula_names' in config:
sel = set(config['formula_names'])
for n, cb in self.index_checkboxes.items(): cb.setChecked(n in sel)
if 'output_file' in config: self.output_file_widget.set_path(config['output_file'])
for name, item in self.index_checkboxes.items():
item.setCheckState(Qt.Checked if name in sel else Qt.Unchecked)
self.enable_checkbox.setChecked(config.get('enabled', True))
if 'output_mode' in config:
btn = self.mode_group.button(config['output_mode'])
if btn:
btn.setChecked(True)
def update_from_config(self, work_dir=None, pipeline=None):
if work_dir: self.work_dir = work_dir
if work_dir:
self.work_dir = work_dir
main = self.window()
if hasattr(main, 'step5_panel'):
p5 = main.step5_panel.output_file.get_path() # 修正：变量名对齐
p5 = main.step5_panel.output_file.get_path()
if p5:
if not os.path.isabs(p5): p5 = os.path.join(self.work_dir or '', p5).replace('\\', '/')
if not os.path.isabs(p5):
p5 = os.path.join(self.work_dir or '', p5)
p5 = p5.replace('\\', '/')
self.training_data_widget.set_path(p5)
def _get_work_dir(self) -> Optional[str]:
if self.work_dir:
out = os.path.join(self.work_dir, "6_water_quality_indices", "training_spectra_indices.csv").replace('\\', '/')
self.output_file_widget.set_path(out)
return self.work_dir
main = self.window()
if hasattr(main, 'work_dir') and main.work_dir:
return main.work_dir
return None
def _get_coord_cols(self, df: pd.DataFrame) -> Tuple[str, str]:
coord_candidates = ['lon', 'lng', 'longitude', '经度', 'x', 'lon_utm', 'utm_x', 'pixel_x']
lat_candidates = ['lat', 'latitude', '纬度', 'y', 'lat_utm', 'utm_y', 'pixel_y']
x_col, y_col = None, None
for col in df.columns:
cl = col.lower()
if x_col is None and any(c in cl for c in coord_candidates):
x_col = col
if y_col is None and any(c in cl for c in lat_candidates):
y_col = col
if x_col is None and len(df.columns) >= 2:
x_col = df.columns[0]
if y_col is None and len(df.columns) >= 2:
y_col = df.columns[1]
return x_col or 'x_coord', y_col or 'y_coord'
def run_step(self):
config = self.get_config()
if not config['training_csv_path']:
QMessageBox.warning(self, "提示", "请先选择输入数据")
if not config['enabled']:
QMessageBox.information(self, "提示", "已禁用计算流程（启用计算流程未勾选）")
return
parent = self.parent()
while parent and not hasattr(parent, 'run_single_step'): parent = parent.parent()
if parent: parent.run_single_step('step8', {'step8': config})
training_path = config['training_csv_path']
if not training_path or not os.path.exists(training_path):
QMessageBox.warning(self, "提示", "请先选择输入特征提取CSV文件")
return
formula_names = config['formula_names']
if not formula_names:
QMessageBox.warning(self, "提示", "请至少勾选一个公式")
return
output_mode = config['output_mode']
try:
from src.utils.water_index import WaterQualityIndexCalculator
calculator = WaterQualityIndexCalculator(self.builtin_formula_path)
spec_df = pd.read_csv(training_path)
x_col, y_col = self._get_coord_cols(spec_df)
results_df = calculator.calculate_many(formula_names, spec_df)
output_df = pd.concat([spec_df, results_df], axis=1)
work_dir = self._get_work_dir()
track_a_path = None
track_b_dir = None
if output_mode in (0, 1):
track_a_dir = os.path.join(work_dir, "6_water_quality_indices") if work_dir else "6_water_quality_indices"
os.makedirs(track_a_dir, exist_ok=True)
track_a_path = os.path.join(track_a_dir, "training_spectra_indices.csv")
if output_mode in (0, 2):
track_b_dir = os.path.join(work_dir, "11_12_13_predictions", "Traditional_Indices") if work_dir else "11_12_13_predictions/Traditional_Indices"
os.makedirs(track_b_dir, exist_ok=True)
saved = []
if output_mode in (0, 1):
output_df.to_csv(track_a_path, index=False, float_format='%.6f')
saved.append(f"宽表: {track_a_path}")
if output_mode in (0, 2):
coord_x = spec_df[x_col].values if x_col in spec_df.columns else np.arange(len(spec_df))
coord_y = spec_df[y_col].values if y_col in spec_df.columns else np.zeros(len(spec_df))
for formula_name in formula_names:
if formula_name not in results_df.columns:
continue
single_df = pd.DataFrame({
'x_coord': coord_x,
'y_coord': coord_y,
'value': results_df[formula_name].values,
})
safe_name = formula_name.replace('/', '_').replace(' ', '_')
out_path = os.path.join(track_b_dir, f"{safe_name}_prediction.csv")
single_df.to_csv(out_path, index=False, float_format='%.6f')
saved.append(f"单文件目录: {track_b_dir}")
QMessageBox.information(
self, "计算完成",
f"已保存 {len(saved)} 个输出目标:\n" + "\n".join(saved)
)
except ImportError as e:
QMessageBox.critical(self, "依赖错误", f"无法导入 WaterQualityIndexCalculator:\n{e}")
except Exception as e:
import traceback
QMessageBox.critical(self, "计算失败", f"原因: {str(e)}\n{traceback.format_exc()}")

19
src/gui/panels/step9_panel.py
View File

@ -315,6 +315,25 @@ class Step9Panel(QWidget):
if not existing or not existing.strip():
self.csv_file.set_path(step5_output_path)
# 1.5 自动探测并回填 Step 8 双轨输出的 Traditional_Indices 目录
if self.work_dir:
trad_indices_dir = os.path.join(
self.work_dir, "11_12_13_predictions", "Traditional_Indices"
)
if os.path.isdir(trad_indices_dir):
csv_files = [
f for f in os.listdir(trad_indices_dir)
if f.lower().endswith('.csv')
]
if csv_files:
csv_files.sort()
first_csv = os.path.join(trad_indices_dir, csv_files[0])
existing = self.csv_file.get_path()
if not existing or not existing.strip():
self.csv_file.set_path(first_csv)
self.refresh_csv_columns()
print(f"✅ 自动探测到 Traditional_Indices 目录，加载首个CSV: {csv_files[0]}")
# 2. 自动填充输出目录（9_Custom_Regression_Modeling）
if self.work_dir:
output_dir = os.path.join(self.work_dir, "9_Custom_Regression_Modeling")