diff --git a/src/gui/panels/step7_index_panel.py b/src/gui/panels/step7_index_panel.py index 0bf8191..b561fcd 100644 --- a/src/gui/panels/step7_index_panel.py +++ b/src/gui/panels/step7_index_panel.py @@ -7,9 +7,8 @@ Step7 面板 - 水质指数计算 import os import sys import pandas as pd -import numpy as np from pathlib import Path -from typing import Dict, List, Optional, Tuple +from typing import Dict, List, Optional from PyQt5.QtWidgets import ( QWidget, QVBoxLayout, QGroupBox, QGridLayout, @@ -146,6 +145,33 @@ class Step7IndexPanel(QWidget): mode_layout.addStretch() output_layout.addLayout(mode_layout) + # ---------- RadioButton 实心美化样式 ---------- + radio_style = """ + QRadioButton { + font-size: 13px; + spacing: 8px; + color: #333333; + } + QRadioButton::indicator { + width: 16px; + height: 16px; + border: 2px solid #999999; + border-radius: 3px; /* 3px是微圆角方形，如果想要纯圆形请改成 8px */ + background-color: white; + } + QRadioButton::indicator:checked { + border: 2px solid #0078d4; + background-color: #0078d4; /* 纯蓝实心 */ + image: none; + } + QRadioButton::indicator:hover { + border: 2px solid #005a9e; + } + """ + self.radio_both.setStyleSheet(radio_style) + self.radio_wide.setStyleSheet(radio_style) + self.radio_single.setStyleSheet(radio_style) + self.enable_checkbox = QCheckBox("启用计算流程") self.enable_checkbox.setChecked(True) output_layout.addWidget(self.enable_checkbox) @@ -261,25 +287,28 @@ class Step7IndexPanel(QWidget): for item in self.index_checkboxes.values(): item.setCheckState(Qt.Unchecked) - def get_config(self) -> Dict: + def get_config(self) -> dict: + """获取配置""" selected = [ name for name, item in self.index_checkboxes.items() if item.checkState() == Qt.Checked ] - # Build coefficient dict for selected formulas - formula_coefficients = { - name: self._formula_coef_map.get(name, []) - for name in selected - } - return { + config = { 'training_csv_path': self.training_data_widget.get_path(), 'formula_csv_file': self.builtin_formula_path, 'formula_names': selected, - 'formula_coefficients': formula_coefficients, 'enabled': self.enable_checkbox.isChecked(), 'output_mode': self.mode_group.checkedId(), } + work_dir = self._get_work_dir() + if work_dir: + track_a_dir = os.path.join(work_dir, "6_water_quality_indices") + os.makedirs(track_a_dir, exist_ok=True) + config['output_file'] = os.path.join(track_a_dir, "training_spectra_indices.csv").replace('\\', '/') + + return config + def set_config(self, config: Dict): if 'training_csv_path' in config: self.training_data_widget.set_path(config['training_csv_path']) @@ -313,117 +342,24 @@ class Step7IndexPanel(QWidget): 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): + """独立运行步骤7 (通过标准的 Worker 路由下发)""" config = self.get_config() if not config['enabled']: QMessageBox.information(self, "提示", "已禁用计算流程（启用计算流程未勾选）") return - training_path = config['training_csv_path'] + training_path = config.get('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: + if not config.get('formula_names'): QMessageBox.warning(self, "提示", "请至少勾选一个公式") return - output_mode = config['output_mode'] - - try: - from src.core.steps.data_preparation_step import DataPreparationStep - - spec_df = pd.read_csv(training_path) - x_col, y_col = self._get_coord_cols(spec_df) - - # 构建 formula_csv_path（使用内置 waterindex.csv） - formula_csv_path = self.builtin_formula_path - if not formula_csv_path or not os.path.exists(formula_csv_path): - # 尝试从 src/gui/model/ 目录找 - possible_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'gui', 'model', 'waterindex.csv') - if os.path.exists(possible_path): - formula_csv_path = possible_path - - work_dir = self._get_work_dir() - - # 调用 DataPreparationStep 的静态方法计算水质指数（宽表输出） - indices_csv_path = DataPreparationStep.calculate_water_quality_indices( - training_csv_path=training_path, - formula_csv_file=formula_csv_path, - formula_names=formula_names, - output_file=None, # 不在此处指定输出，由下面的双轨输出逻辑接管 - enabled=True, - output_dir=work_dir if work_dir else os.getcwd(), - ) - - # 读取计算结果（宽表） - if indices_csv_path and os.path.exists(indices_csv_path): - output_df = pd.read_csv(indices_csv_path) - else: - output_df = spec_df # fallback - - 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 output_df.columns: - continue - single_df = pd.DataFrame({ - 'x_coord': coord_x, - 'y_coord': coord_y, - 'value': output_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"无法导入模块:\n{e}") - except Exception as e: - import traceback - QMessageBox.critical(self, "计算失败", f"原因: {str(e)}\n{traceback.format_exc()}") \ No newline at end of file + main_window = self.window() + if hasattr(main_window, 'run_single_step'): + pipeline_config = {'step7_index': config} + main_window.run_single_step('step7_index', pipeline_config) \ No newline at end of file