Compare commits

...

2 Commits

2 changed files with 138 additions and 329 deletions

View File

@ -141,3 +141,8 @@ class FileSelectWidget(QWidget):
def set_path(self, path): def set_path(self, path):
"""设置路径""" """设置路径"""
self.line_edit.setText(str(path)) self.line_edit.setText(str(path))
def set_read_only(self, read_only=True):
"""设置文件选择框为只读,并禁用浏览按钮。"""
self.line_edit.setReadOnly(read_only)
self.browse_btn.setEnabled(not read_only)

View File

@ -1,408 +1,212 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Step5_5 面板 - 水质指数计算
"""
import os import os
import sys import sys
import pandas as pd
from pathlib import Path from pathlib import Path
from typing import Dict, List, Union from typing import Dict, List, Union
def get_resource_path(relative_path: str) -> str:
"""获取资源的绝对路径,适配 PyInstaller 打包环境。"""
if hasattr(sys, '_MEIPASS'):
return os.path.join(sys._MEIPASS, relative_path)
return os.path.abspath(
os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), relative_path)
)
import pandas as pd
from PyQt5.QtWidgets import ( from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QGroupBox, QFormLayout, QGridLayout, QWidget, QVBoxLayout, QGroupBox, QGridLayout,
QHBoxLayout, QLabel, QLineEdit, QComboBox, QCheckBox, QHBoxLayout, QLabel, QCheckBox, QPushButton, QMessageBox, QScrollArea
QPushButton, QMessageBox,
) )
from PyQt5.QtCore import Qt from PyQt5.QtCore import Qt
from src.gui.components.custom_widgets import FileSelectWidget from src.gui.components.custom_widgets import FileSelectWidget
from src.gui.styles import ModernStylesheet from src.gui.styles import ModernStylesheet
def get_resource_path(relative_path: str) -> str:
"""适配开发与 PyInstaller 环境的路径获取逻辑"""
if hasattr(sys, '_MEIPASS'):
# 打包后,文件会被平铺或按 tree 结构放入临时目录
return os.path.join(sys._MEIPASS, relative_path)
# 开发环境下:基于当前文件 (step5_5_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)
class Step5_5Panel(QWidget): class Step5_5Panel(QWidget):
"""步骤5.5:水质指数计算"""
def __init__(self, parent=None): def __init__(self, parent=None):
super().__init__(parent) super().__init__(parent)
self.index_checkboxes: Dict[str, QCheckBox] = {} self.index_checkboxes: Dict[str, QCheckBox] = {}
self.csv_columns = [] # 存储CSV文件列名 # 标识为 waterindex.csv目录跳转逻辑在 get_resource_path 中
self.builtin_formula_path = get_resource_path("waterindex.csv")
self.init_ui() self.init_ui()
# 延迟一小会儿加载确保UI框架已就绪
self._auto_load_formulas()
def init_ui(self): def init_ui(self):
main_layout = QVBoxLayout() main_layout = QVBoxLayout()
main_layout.setContentsMargins(20, 20, 20, 20)
main_layout.setSpacing(10)
# 标题 # 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)
main_layout.addWidget(path_group)
# 2. 训练数据输入
input_group = QGroupBox("输入样本数据")
input_layout = QVBoxLayout()
self.training_data_widget = FileSelectWidget("特征提取CSV:", "CSV Files (*.csv)")
input_layout.addWidget(self.training_data_widget)
input_group.setLayout(input_layout)
main_layout.addWidget(input_group)
# 数据文件选择 # 3. 公式选择
data_group = QGroupBox("数据文件") self.formula_group = QGroupBox("待计算水质指数勾选")
data_layout = QVBoxLayout()
# 训练数据CSV文件选择
self.training_data_widget = FileSelectWidget("训练数据CSV文件:", "CSV Files (*.csv)")
data_layout.addWidget(self.training_data_widget)
# 公式CSV文件选择
self.formula_csv_widget = FileSelectWidget("公式CSV文件:", "CSV Files (*.csv)")
data_layout.addWidget(self.formula_csv_widget)
# 刷新公式按钮
refresh_layout = QHBoxLayout()
self.refresh_button = QPushButton("刷新公式列表")
self.refresh_button.clicked.connect(self.refresh_formulas)
refresh_layout.addWidget(self.refresh_button)
refresh_layout.addStretch()
data_layout.addLayout(refresh_layout)
data_group.setLayout(data_layout)
main_layout.addWidget(data_group)
# 公式选择区域
self.formula_group = QGroupBox("选择要计算的公式")
formula_outer_layout = QVBoxLayout() formula_outer_layout = QVBoxLayout()
# 按钮控制区域 btn_layout = QHBoxLayout()
button_layout = QHBoxLayout()
self.select_all_btn = QPushButton("全选") self.select_all_btn = QPushButton("全选")
self.select_all_btn.clicked.connect(self.select_all_formulas)
self.deselect_all_btn = QPushButton("清空") self.deselect_all_btn = QPushButton("清空")
self.select_all_btn.clicked.connect(self.select_all_formulas)
self.deselect_all_btn.clicked.connect(self.deselect_all_formulas) self.deselect_all_btn.clicked.connect(self.deselect_all_formulas)
button_layout.addWidget(self.select_all_btn) btn_layout.addWidget(self.select_all_btn)
button_layout.addWidget(self.deselect_all_btn) btn_layout.addWidget(self.deselect_all_btn)
button_layout.addStretch() btn_layout.addStretch()
formula_outer_layout.addLayout(button_layout) 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)
self.formula_layout = QGridLayout()
formula_outer_layout.addLayout(self.formula_layout) # 核心滚动区
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setMinimumHeight(300) # 强制最小高度,防止塌陷
self.scroll_content = QWidget()
self.formula_layout = QGridLayout(self.scroll_content)
self.formula_layout.setAlignment(Qt.AlignTop) # 靠顶对齐
scroll.setWidget(self.scroll_content)
formula_outer_layout.addWidget(scroll)
self.formula_group.setLayout(formula_outer_layout) self.formula_group.setLayout(formula_outer_layout)
main_layout.addWidget(self.formula_group) main_layout.addWidget(self.formula_group)
# 输出文件设置 # 4. 输出与运行
output_group = QGroupBox("输出设置") output_group = QGroupBox("结果输出")
output_layout = QVBoxLayout() output_layout = QVBoxLayout()
self.output_file_widget = FileSelectWidget("保存路径:", "CSV Files (*.csv)", mode="save")
self.output_file_widget = FileSelectWidget(
"输出文件:", "CSV Files (*.csv)", mode="save"
)
output_layout.addWidget(self.output_file_widget) output_layout.addWidget(self.output_file_widget)
output_group.setLayout(output_layout) output_group.setLayout(output_layout)
main_layout.addWidget(output_group) main_layout.addWidget(output_group)
# 启用选项 self.enable_checkbox = QCheckBox("启用计算流程")
self.enable_checkbox = QCheckBox("启用此步骤")
self.enable_checkbox.setChecked(True) self.enable_checkbox.setChecked(True)
main_layout.addWidget(self.enable_checkbox) main_layout.addWidget(self.enable_checkbox)
# 独立运行按钮 self.run_button = QPushButton("立即执行计算")
self.run_button = QPushButton("独立运行此步骤")
self.run_button.setStyleSheet(ModernStylesheet.get_button_stylesheet('success')) self.run_button.setStyleSheet(ModernStylesheet.get_button_stylesheet('success'))
self.run_button.setMinimumHeight(40)
self.run_button.clicked.connect(self.run_step) self.run_button.clicked.connect(self.run_step)
main_layout.addWidget(self.run_button) main_layout.addWidget(self.run_button)
# 公式编辑区域
formula_edit_group = QGroupBox("添加自定义公式")
formula_edit_layout = QFormLayout()
self.formula_name_edit = QLineEdit()
# 公式类别下拉选择框
self.formula_category_combo = QComboBox()
self.formula_category_combo.addItems([
"chlorophyll_a",
"Phycocyanin (BGA_PC)",
"Total Nitrogen (TN)",
"Total Phosphorus (TP)",
"Orthophosphate",
"COD",
"BOD",
"TOC",
"Dissolved Oxygen (DO)",
"E. coli",
"Total Coliforms",
"Turbidity",
"Total Suspended Solids (TSS)",
"Color",
"pH",
"Temperature",
"Conductivity",
"Total Dissolved Solids (TDS)"
])
self.formula_category_combo.setEditable(True) # 允许用户输入自定义类别
self.formula_expression_edit = QLineEdit()
self.formula_reference_edit = QLineEdit()
formula_edit_layout.addRow("公式名称:", self.formula_name_edit)
formula_edit_layout.addRow("公式类别:", self.formula_category_combo)
formula_edit_layout.addRow("公式表达式:", self.formula_expression_edit)
formula_edit_layout.addRow("参考文献:", self.formula_reference_edit)
add_button = QPushButton("添加公式")
add_button.clicked.connect(self.add_custom_formula)
formula_edit_layout.addRow(add_button)
formula_edit_group.setLayout(formula_edit_layout)
main_layout.addWidget(formula_edit_group)
main_layout.addStretch()
self.setLayout(main_layout) self.setLayout(main_layout)
# 自动加载内置公式文件 def _auto_load_formulas(self):
formula_csv_path = get_resource_path("data/sub/waterindex.csv") """启动时自动加载逻辑"""
if os.path.isfile(formula_csv_path): if os.path.exists(self.builtin_formula_path):
self.formula_csv_widget.set_path(str(formula_csv_path)) self.refresh_formulas(silent=True)
self.refresh_formulas() else:
print(f"DEBUG: 自动加载失败,路径不存在: {self.builtin_formula_path}")
def refresh_formulas(self): def refresh_formulas(self, silent=False):
"""刷新公式列表""" path = self.builtin_formula_path
formula_csv_path = self.formula_csv_widget.get_path() if not os.path.exists(path):
if not formula_csv_path or not os.path.exists(formula_csv_path): if not silent: QMessageBox.warning(self, "错误", f"找不到内置公式文件:\n{path}")
QMessageBox.warning(self, "警告", "请先选择有效的公式CSV文件")
return return
try: try:
# 清除现有的勾选框 # 清理旧列表
for checkbox in self.index_checkboxes.values(): for i in reversed(range(self.formula_layout.count())):
self.formula_layout.removeWidget(checkbox) widget = self.formula_layout.itemAt(i).widget()
checkbox.deleteLater() if widget: widget.deleteLater()
self.index_checkboxes.clear() self.index_checkboxes.clear()
# 读取公式CSV文件 # 鲁棒性读取:尝试不同编码
df = pd.read_csv(formula_csv_path) for encoding in ['utf-8', 'gbk', 'utf-8-sig']:
if df.empty or 'Formula_Name' not in df.columns: try:
QMessageBox.warning(self, "警告", "公式CSV文件格式不正确") df = pd.read_csv(path, encoding=encoding)
if 'Formula_Name' in df.columns: break
except: continue
if 'Formula_Name' not in df.columns:
if not silent: QMessageBox.critical(self, "错误", "CSV文件缺少 'Formula_Name'")
return return
# 获取所有公式名称(跳过第一行) names = df['Formula_Name'].dropna().unique().tolist()
formula_names = df['Formula_Name'].tolist()[1:]
# 创建3列布局的勾选框
row, col = 0, 0 row, col = 0, 0
for formula_name in formula_names: for name in names:
if pd.isna(formula_name) or not formula_name.strip(): name = str(name).strip()
continue if not name: continue
cb = QCheckBox(name)
checkbox = QCheckBox(formula_name.strip()) cb.setChecked(True)
checkbox.setChecked(True) self.index_checkboxes[name] = cb
self.index_checkboxes[formula_name.strip()] = checkbox self.formula_layout.addWidget(cb, row, col)
self.formula_layout.addWidget(checkbox, row, col)
col += 1 col += 1
if col >= 3: # 每行3列 if col >= 3:
col = 0 col = 0
row += 1 row += 1
except Exception as e: # 强制UI更新
QMessageBox.critical(self, "错误", f"读取公式文件失败: {str(e)}") self.scroll_content.adjustSize()
print(f"✅ 成功加载 {len(self.index_checkboxes)} 个公式")
def add_custom_formula(self):
"""添加自定义公式到公式CSV文件"""
formula_csv_path = self.formula_csv_widget.get_path()
if not formula_csv_path:
QMessageBox.warning(self, "警告", "请先选择公式CSV文件")
return
formula_name = self.formula_name_edit.text().strip()
formula_category = self.formula_category_combo.currentText().strip()
formula_expression = self.formula_expression_edit.text().strip()
formula_reference = self.formula_reference_edit.text().strip()
if not all([formula_name, formula_category, formula_expression]):
QMessageBox.warning(self, "警告", "请填写公式名称、类别和表达式")
return
try:
# 读取现有公式文件或创建新文件
if os.path.exists(formula_csv_path):
df = pd.read_csv(formula_csv_path)
else:
df = pd.DataFrame(columns=['Formula_Name', 'Category', 'Formula', 'Reference'])
# 添加新公式
new_row = pd.DataFrame({
'Formula_Name': [formula_name],
'Category': [formula_category],
'Formula': [formula_expression],
'Reference': [formula_reference]
})
df = pd.concat([df, new_row], ignore_index=True)
# 保存文件
df.to_csv(formula_csv_path, index=False, encoding='utf-8')
# 清空输入框
self.formula_name_edit.clear()
self.formula_category_combo.setCurrentIndex(0) # 重置到第一个选项
self.formula_expression_edit.clear()
self.formula_reference_edit.clear()
# 刷新公式列表
self.refresh_formulas()
QMessageBox.information(self, "成功", "公式添加成功")
except Exception as e: except Exception as e:
QMessageBox.critical(self, "错误", f"添加公式失败: {str(e)}") if not silent: QMessageBox.critical(self, "加载失败", f"原因: {str(e)}")
def get_config(self) -> Dict[str, Union[List[str], str, bool]]: def select_all_formulas(self):
"""获取配置""" for cb in self.index_checkboxes.values(): cb.setChecked(True)
selected = [
name for name, checkbox in self.index_checkboxes.items() def deselect_all_formulas(self):
if checkbox.isChecked() for cb in self.index_checkboxes.values(): cb.setChecked(False)
]
output_path = self.output_file_widget.get_path() def get_config(self):
selected = [n for n, cb in self.index_checkboxes.items() if cb.isChecked()]
return { return {
'training_spectra_path': self.training_data_widget.get_path() or None, 'training_spectra_path': self.training_data_widget.get_path(),
'formula_csv_file': self.formula_csv_widget.get_path() or None, 'formula_csv_file': self.builtin_formula_path,
'formula_names': selected, 'formula_names': selected,
'output_file': output_path or None, 'output_file': self.output_file_widget.get_path(),
'enabled': self.enable_checkbox.isChecked() 'enabled': self.enable_checkbox.isChecked()
} }
def set_config(self, config): def set_config(self, config):
"""设置配置""" if 'training_spectra_path' in config: self.training_data_widget.set_path(config['training_spectra_path'])
if 'training_spectra_path' in config:
self.training_data_widget.set_path(config['training_spectra_path'])
if 'formula_csv_file' in config:
self.formula_csv_widget.set_path(config['formula_csv_file'])
self.refresh_formulas()
if 'formula_names' in config: if 'formula_names' in config:
selected_formulas = set(config['formula_names']) sel = set(config['formula_names'])
for name, checkbox in self.index_checkboxes.items(): for n, cb in self.index_checkboxes.items(): cb.setChecked(n in sel)
checkbox.setChecked(name in selected_formulas) if 'output_file' in config: self.output_file_widget.set_path(config['output_file'])
self.enable_checkbox.setChecked(config.get('enabled', True))
if 'output_file' in config and config['output_file']:
self.output_file_widget.set_path(config['output_file'])
elif 'output_filename' in config and config['output_filename']:
self.output_file_widget.set_path(config['output_filename'])
if 'enabled' in config:
self.enable_checkbox.setChecked(config['enabled'])
def update_from_config(self, work_dir=None, pipeline=None): def update_from_config(self, work_dir=None, pipeline=None):
"""从全局配置自动填充训练数据和输出路径 if work_dir: self.work_dir = work_dir
main = self.window()
if hasattr(main, 'step5_panel'):
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('\\', '/')
self.training_data_widget.set_path(p5)
Args:
work_dir: 工作目录路径
pipeline: Pipeline 实例(未使用,保留接口兼容性)
"""
if work_dir:
self.work_dir = work_dir
elif hasattr(self, 'work_dir') and self.work_dir:
pass
else:
self.work_dir = None
# 1. 自动填入训练数据路径(从 Step5 的输出中获取)
# 优先级:直接 widget > pipeline.step_outputs 回退
main_window = self.window()
if hasattr(main_window, 'step5_panel'):
# 优先直接从 Step5 的输出 widget 读取(已运行的最新输出)
step5_output = main_window.step5_panel.output_file.get_path()
if step5_output:
# 若为相对路径,使用 work_dir 合成为绝对路径
if not os.path.isabs(step5_output):
step5_output = os.path.join(self.work_dir or '', step5_output).replace('\\', '/')
self.training_data_widget.set_path(step5_output)
else:
# 退而求其次,使用 Step5 的输入 CSV
step5_csv = main_window.step5_panel.csv_file.get_path()
if step5_csv:
# 若为相对路径,使用 work_dir 合成为绝对路径
if not os.path.isabs(step5_csv):
step5_csv = os.path.join(self.work_dir or '', step5_csv).replace('\\', '/')
self.training_data_widget.set_path(step5_csv)
# 如果上述都没找到,尝试从 pipeline.step_outputs 回退
if not self.training_data_widget.get_path() and pipeline and hasattr(pipeline, 'step_outputs'):
step5_outputs = getattr(pipeline, 'step_outputs', {}).get('step5', {})
training_path = step5_outputs.get('training_spectra')
if training_path:
self.training_data_widget.set_path(training_path)
# 2. 自动填入输出文件的绝对路径
if self.work_dir: if self.work_dir:
output_abs = os.path.join(self.work_dir, "6_water_quality_indices", out = os.path.join(self.work_dir, "6_water_quality_indices", "training_spectra_indices.csv").replace('\\', '/')
"training_spectra_indices.csv").replace('\\', '/') self.output_file_widget.set_path(out)
self.output_file_widget.set_path(output_abs)
def is_enabled(self) -> bool:
return self.enable_checkbox.isChecked()
def select_all_formulas(self):
"""全选所有公式"""
for checkbox in self.index_checkboxes.values():
checkbox.setChecked(True)
def deselect_all_formulas(self):
"""清空所有公式"""
for checkbox in self.index_checkboxes.values():
checkbox.setChecked(False)
def run_step(self): def run_step(self):
"""独立运行步骤5.5:计算水质指数。
动态根据输入 CSV 文件名生成输出文件名,自动填入 output_file_widget。
例如training_spectra.csv → training_spectra_indices.csv
sampling_spectra.csv → sampling_spectra_indices.csv
"""
# 验证输入
training_csv_path = self.training_data_widget.get_path()
formula_csv_path = self.formula_csv_widget.get_path()
if not training_csv_path:
QMessageBox.warning(self, "输入验证失败", "请选择训练数据CSV文件")
return
if not formula_csv_path:
QMessageBox.warning(self, "输入验证失败", "请选择公式CSV文件")
return
if not os.path.exists(training_csv_path):
QMessageBox.warning(self, "输入验证失败", "训练数据CSV文件不存在")
return
if not os.path.exists(formula_csv_path):
QMessageBox.warning(self, "输入验证失败", "公式CSV文件不存在")
return
# 动态生成输出文件:自动拼接 _indices 后缀
input_name = Path(training_csv_path).stem
dynamic_output = f"{input_name}_indices.csv"
# 合成完整绝对路径(优先使用 work_dir其次从 training_csv_path 推导)
work_dir = getattr(self, 'work_dir', None)
if work_dir:
dynamic_output = os.path.join(
work_dir, "6_water_quality_indices", dynamic_output
).replace('\\', '/')
self.output_file_widget.set_path(dynamic_output)
# 获取配置
config = self.get_config() config = self.get_config()
if not config['training_spectra_path']:
# 调用GUI的run_single_step方法 QMessageBox.warning(self, "提示", "请先选择输入数据")
return
parent = self.parent() parent = self.parent()
while parent and not hasattr(parent, 'run_single_step'): while parent and not hasattr(parent, 'run_single_step'): parent = parent.parent()
parent = parent.parent() if parent: parent.run_single_step('step5_5', {'step5_5': config})
if parent and hasattr(parent, 'run_single_step'):
parent.run_single_step('step5_5', {'step5_5': config})
else:
QMessageBox.critical(self, "错误", "无法找到父级GUI对象")