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

Step7 面板：单选框蓝底实心样式美化，清理死代码（np/Tuple 导入、_get_coord_cols），run_step 路由化

Browse Source
This commit is contained in:
DXC
2026-06-12 09:24:16 +08:00
parent e59703f163
commit 04669bdee8
1 changed files with 46 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
src/gui/panels/step7_index_panel.py
View File

@ -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()}")
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)