diff --git a/src/gui/components/custom_widgets.py b/src/gui/components/custom_widgets.py index 0df2b42..6cc7557 100644 --- a/src/gui/components/custom_widgets.py +++ b/src/gui/components/custom_widgets.py @@ -140,4 +140,9 @@ class FileSelectWidget(QWidget): def set_path(self, path): """设置路径""" - self.line_edit.setText(str(path)) \ No newline at end of file + 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) \ No newline at end of file diff --git a/src/gui/panels/step5_5_panel.py b/src/gui/panels/step5_5_panel.py index 89427c0..0f24462 100644 --- a/src/gui/panels/step5_5_panel.py +++ b/src/gui/panels/step5_5_panel.py @@ -1,408 +1,171 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -Step5_5 面板 - 水质指数计算 -""" - import os import sys +import pandas as pd from pathlib import Path 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 ( - QWidget, QVBoxLayout, QGroupBox, QFormLayout, QGridLayout, - QHBoxLayout, QLabel, QLineEdit, QComboBox, QCheckBox, - QPushButton, QMessageBox, + QWidget, QVBoxLayout, QGroupBox, QGridLayout, + QHBoxLayout, QLabel, QCheckBox, QPushButton, QMessageBox, QScrollArea ) from PyQt5.QtCore import Qt - from src.gui.components.custom_widgets import FileSelectWidget from src.gui.styles import ModernStylesheet +def get_resource_path(relative_path: str) -> str: + 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)) class Step5_5Panel(QWidget): - """步骤5.5：水质指数计算""" - def __init__(self, parent=None): super().__init__(parent) self.index_checkboxes: Dict[str, QCheckBox] = {} - self.csv_columns = [] # 存储CSV文件列名 + self.builtin_formula_path = get_resource_path("data/sub/waterindex.csv") self.init_ui() + self._auto_load_formulas() def init_ui(self): main_layout = QVBoxLayout() + main_layout.setContentsMargins(20, 20, 20, 20) - # 标题 - - - # 数据文件选择 - data_group = QGroupBox("数据文件") + # 1. 数据文件（隐藏公式路径，自动填入训练数据） + data_group = QGroupBox("输入数据") data_layout = QVBoxLayout() - - # 训练数据CSV文件选择 - self.training_data_widget = FileSelectWidget("训练数据CSV文件:", "CSV Files (*.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)") + self.formula_csv_widget = FileSelectWidget("公式配置:", "CSV Files (*.csv)") + self.formula_csv_widget.hide() # 界面隐身 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("选择要计算的公式") + # 2. 公式选择区 + self.formula_group = QGroupBox("水质指数公式勾选") formula_outer_layout = QVBoxLayout() - # 按钮控制区域 - button_layout = QHBoxLayout() + btn_layout = QHBoxLayout() self.select_all_btn = QPushButton("全选") - self.select_all_btn.clicked.connect(self.select_all_formulas) 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) - button_layout.addWidget(self.select_all_btn) - button_layout.addWidget(self.deselect_all_btn) - button_layout.addStretch() + btn_layout.addWidget(self.select_all_btn) + btn_layout.addWidget(self.deselect_all_btn) + btn_layout.addStretch() + formula_outer_layout.addLayout(btn_layout) - formula_outer_layout.addLayout(button_layout) - - # 公式勾选框网格布局 - self.formula_layout = QGridLayout() - formula_outer_layout.addLayout(self.formula_layout) + # 滚动显示区域 + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll_content = QWidget() + self.formula_layout = QGridLayout(scroll_content) + scroll.setWidget(scroll_content) + formula_outer_layout.addWidget(scroll) self.formula_group.setLayout(formula_outer_layout) main_layout.addWidget(self.formula_group) - # 输出文件设置 + # 3. 输出设置 output_group = QGroupBox("输出设置") output_layout = QVBoxLayout() - - self.output_file_widget = FileSelectWidget( - "输出文件:", "CSV Files (*.csv)", mode="save" - ) + self.output_file_widget = FileSelectWidget("输出CSV路径:", "CSV Files (*.csv)", mode="save") output_layout.addWidget(self.output_file_widget) - output_group.setLayout(output_layout) main_layout.addWidget(output_group) - # 启用选项 + # 4. 操作区 self.enable_checkbox = QCheckBox("启用此步骤") self.enable_checkbox.setChecked(True) main_layout.addWidget(self.enable_checkbox) - # 独立运行按钮 self.run_button = QPushButton("独立运行此步骤") self.run_button.setStyleSheet(ModernStylesheet.get_button_stylesheet('success')) self.run_button.clicked.connect(self.run_step) 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) - # 自动加载内置公式文件 - formula_csv_path = get_resource_path("data/sub/waterindex.csv") - if os.path.isfile(formula_csv_path): - self.formula_csv_widget.set_path(str(formula_csv_path)) - self.refresh_formulas() + def _auto_load_formulas(self): + if os.path.isfile(self.builtin_formula_path): + self.formula_csv_widget.set_path(self.builtin_formula_path) + self.refresh_formulas(silent=True) - def refresh_formulas(self): - """刷新公式列表""" - formula_csv_path = self.formula_csv_widget.get_path() - if not formula_csv_path or not os.path.exists(formula_csv_path): - QMessageBox.warning(self, "警告", "请先选择有效的公式CSV文件") + def refresh_formulas(self, silent=False): + path = self.formula_csv_widget.get_path() + if not path or not os.path.exists(path): + if not silent: QMessageBox.warning(self, "警告", "内置公式文件丢失") return - try: - # 清除现有的勾选框 - for checkbox in self.index_checkboxes.values(): - self.formula_layout.removeWidget(checkbox) - checkbox.deleteLater() + # 清理旧项 + for i in reversed(range(self.formula_layout.count())): + self.formula_layout.itemAt(i).widget().setParent(None) self.index_checkboxes.clear() - # 读取公式CSV文件 - df = pd.read_csv(formula_csv_path) - if df.empty or 'Formula_Name' not in df.columns: - QMessageBox.warning(self, "警告", "公式CSV文件格式不正确") - return + df = pd.read_csv(path) + # 修正：不使用 [1:] 切片，直接读取所有有效行 + formula_names = df['Formula_Name'].dropna().unique().tolist() - # 获取所有公式名称（跳过第一行） - formula_names = df['Formula_Name'].tolist()[1:] - - # 创建3列布局的勾选框 row, col = 0, 0 - for formula_name in formula_names: - if pd.isna(formula_name) or not formula_name.strip(): - continue - - checkbox = QCheckBox(formula_name.strip()) - checkbox.setChecked(True) - self.index_checkboxes[formula_name.strip()] = checkbox - self.formula_layout.addWidget(checkbox, row, col) - + for name in formula_names: + name = name.strip() + cb = QCheckBox(name) + cb.setChecked(True) + self.index_checkboxes[name] = cb + self.formula_layout.addWidget(cb, row, col) col += 1 - if col >= 3: # 每行3列 + if col >= 3: col = 0 row += 1 - except Exception as e: - QMessageBox.critical(self, "错误", f"读取公式文件失败: {str(e)}") + if not silent: QMessageBox.critical(self, "错误", f"解析公式失败: {e}") - 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 + def select_all_formulas(self): + for cb in self.index_checkboxes.values(): cb.setChecked(True) - 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() + def deselect_all_formulas(self): + for cb in self.index_checkboxes.values(): cb.setChecked(False) - 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: - QMessageBox.critical(self, "错误", f"添加公式失败: {str(e)}") - - def get_config(self) -> Dict[str, Union[List[str], str, bool]]: - """获取配置""" - selected = [ - name for name, checkbox in self.index_checkboxes.items() - if checkbox.isChecked() - ] - 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 { - 'training_spectra_path': self.training_data_widget.get_path() or None, - 'formula_csv_file': self.formula_csv_widget.get_path() or None, + 'training_spectra_path': self.training_data_widget.get_path(), + 'formula_csv_file': self.formula_csv_widget.get_path(), 'formula_names': selected, - 'output_file': output_path or None, + 'output_file': self.output_file_widget.get_path(), 'enabled': self.enable_checkbox.isChecked() } 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() - + self.refresh_formulas(silent=True) if 'formula_names' in config: - selected_formulas = set(config['formula_names']) - for name, checkbox in self.index_checkboxes.items(): - checkbox.setChecked(name in selected_formulas) - - 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']) + 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']) + if 'enabled' in config: self.enable_checkbox.setChecked(config['enabled']) 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: - output_abs = os.path.join(self.work_dir, "6_water_quality_indices", - "training_spectra_indices.csv").replace('\\', '/') - 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) + out = os.path.join(self.work_dir, "6_water_quality_indices", "training_spectra_indices.csv").replace('\\', '/') + self.output_file_widget.set_path(out) 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() - - # 调用GUI的run_single_step方法 + if not config['training_spectra_path']: + QMessageBox.warning(self, "提示", "请选择输入数据") + return parent = self.parent() - while parent and not hasattr(parent, 'run_single_step'): - parent = parent.parent() - - if parent and hasattr(parent, 'run_single_step'): - parent.run_single_step('step5_5', {'step5_5': config}) - else: - QMessageBox.critical(self, "错误", "无法找到父级GUI对象") + while parent and not hasattr(parent, 'run_single_step'): parent = parent.parent() + if parent: parent.run_single_step('step5_5', {'step5_5': config}) \ No newline at end of file