#!/usr/bin/env python # -*- coding: utf-8 -*- """ Step4 面板 - 采样点布设 (现代化排版重构) """ import os import sys from pathlib import Path # 路径归一化 helper _HERE = os.path.dirname(os.path.abspath(__file__)) if _HERE not in sys.path: sys.path.insert(0, _HERE) from _step_path_resolver import resolve_subdir from PyQt5.QtCore import QTimer, Qt from PyQt5.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QFormLayout, QPushButton, QCheckBox, QSpinBox, QMessageBox, QLabel, QFrame ) from src.gui.components.custom_widgets import FileSelectWidget from src.gui.dialogs import SamplingViewerDialog from src.gui.styles import ModernStylesheet class Step4SamplingPanel(QWidget): """步骤4：采样点布设""" def __init__(self, parent=None): super().__init__(parent) self.init_ui() def init_ui(self): # 注入全局样式系统 self.setStyleSheet(ModernStylesheet.get_main_stylesheet()) # 主布局：增加四周留白(24px)和模块间的呼吸间距(20px) main_layout = QVBoxLayout() main_layout.setContentsMargins(24, 24, 24, 24) main_layout.setSpacing(20) # ========================================== # 卡片 1：输入数据配置 # ========================================== input_group = QGroupBox("📁 输入数据") input_layout = QVBoxLayout() input_layout.setSpacing(16) input_layout.setContentsMargins(20, 24, 20, 20) self.deglint_img_file = FileSelectWidget( "去耀斑影像:", "Image Files (*.bsq *.dat *.tif);;All Files (*.*)" ) self.water_mask_file = FileSelectWidget( "水域掩膜图:", "Mask Files (*.dat *.tif);;All Files (*.*)" ) input_layout.addWidget(self.deglint_img_file) input_layout.addWidget(self.water_mask_file) input_group.setLayout(input_layout) main_layout.addWidget(input_group) # ========================================== # 卡片 2：核心参数设置 # ========================================== params_group = QGroupBox("⚙️ 采样参数") params_layout = QFormLayout() params_layout.setSpacing(16) params_layout.setContentsMargins(20, 24, 20, 20) self.interval = QSpinBox() self.interval.setRange(10, 500) self.interval.setValue(50) self.interval.setSuffix(" px") self.interval.setMinimumWidth(120) params_layout.addRow("采样点间隔:", self.interval) self.sample_radius = QSpinBox() self.sample_radius.setRange(1, 50) self.sample_radius.setValue(5) self.sample_radius.setSuffix(" px") self.sample_radius.setMinimumWidth(120) params_layout.addRow("中心采样半径:", self.sample_radius) self.chunk_size = QSpinBox() self.chunk_size.setRange(100, 10000) self.chunk_size.setValue(1000) self.chunk_size.setSuffix(" px") self.chunk_size.setMinimumWidth(120) params_layout.addRow("内存处理块大小:", self.chunk_size) self.use_adaptive_sampling = QCheckBox("启用自适应边缘采样") self.use_adaptive_sampling.setChecked(True) params_layout.addRow("智能模式:", self.use_adaptive_sampling) params_group.setLayout(params_layout) main_layout.addWidget(params_group) # ========================================== # 卡片 3：输出与执行 # ========================================== output_group = QGroupBox("🚀 输出与执行") output_layout = QVBoxLayout() output_layout.setSpacing(16) output_layout.setContentsMargins(20, 24, 20, 20) # 输出文件选择 self.output_file = FileSelectWidget( "结果保存至:", "CSV Files (*.csv);;All Files (*.*)", mode="save" ) self.output_file.line_edit.setPlaceholderText("sampling_spectra.csv") output_layout.addWidget(self.output_file) # 底部操作栏 (水平布局) action_layout = QHBoxLayout() self.enable_checkbox = QCheckBox("启用此步骤") self.enable_checkbox.setChecked(True) action_layout.addWidget(self.enable_checkbox) action_layout.addStretch() # 把按钮推到右边 self.preview_btn = QPushButton("交互式预览采样点") self.preview_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('normal')) self.preview_btn.setEnabled(False) self.preview_btn.setMinimumWidth(160) self.preview_btn.clicked.connect(self._open_sampling_viewer) self.run_btn = QPushButton("独立运行步骤") self.run_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('primary')) self.run_btn.setMinimumWidth(140) self.run_btn.clicked.connect(self._on_run_single_clicked) action_layout.addWidget(self.preview_btn) action_layout.addWidget(self.run_btn) output_layout.addLayout(action_layout) output_group.setLayout(output_layout) main_layout.addWidget(output_group) main_layout.addStretch() self.setLayout(main_layout) # 添加心跳定时器 self._status_timer = QTimer(self) self._status_timer.timeout.connect(self._check_csv_exists) self._status_timer.start(2000) # 监听输出路径变化 self.output_file.line_edit.textChanged.connect(self._on_output_changed) # ================= 下方业务逻辑保持不变 ================= def get_config(self): config = { 'interval': self.interval.value(), 'sample_radius': self.sample_radius.value(), 'chunk_size': self.chunk_size.value(), 'use_adaptive_sampling': self.use_adaptive_sampling.isChecked(), } deglint_img_path = self.deglint_img_file.get_path() if deglint_img_path: config['deglint_img_path'] = deglint_img_path water_mask_path = self.water_mask_file.get_path() if water_mask_path: config['water_mask_path'] = water_mask_path return config def set_config(self, config): if 'interval' in config: self.interval.setValue(config['interval']) if 'sample_radius' in config: self.sample_radius.setValue(config['sample_radius']) if 'chunk_size' in config: self.chunk_size.setValue(config['chunk_size']) if 'use_adaptive_sampling' in config: self.use_adaptive_sampling.setChecked(config['use_adaptive_sampling']) if 'deglint_img_path' in config: self.deglint_img_file.set_path(config['deglint_img_path']) if 'water_mask_path' in config: self.water_mask_file.set_path(config['water_mask_path']) def update_from_config(self, work_dir=None, pipeline=None): if work_dir: self.work_dir = work_dir elif hasattr(self, 'work_dir') and self.work_dir: pass else: self.work_dir = None main_window = self.window() deglint_path = None if pipeline and hasattr(pipeline, 'step_outputs'): step3_outputs = getattr(pipeline, 'step_outputs', {}).get('step3', {}) deglint_path = ( step3_outputs.get('deglint_image') or step3_outputs.get('output_path') or step3_outputs.get('output_file') or step3_outputs.get('deglint_img_path') ) if not deglint_path and hasattr(main_window, 'step3_panel'): deglint_path = main_window.step3_panel.output_file.get_path() if deglint_path: if not os.path.isabs(deglint_path): deglint_path = os.path.join(self.work_dir or '', deglint_path).replace('\\', '/') self.deglint_img_file.set_path(deglint_path) water_mask_path = None if pipeline and hasattr(pipeline, 'step_outputs'): step1_outputs = getattr(pipeline, 'step_outputs', {}).get('step1', {}) water_mask_path = ( step1_outputs.get('water_mask') or step1_outputs.get('output_path') or step1_outputs.get( 'output_file') ) if not water_mask_path and hasattr(main_window, 'step1_panel'): water_mask_path = main_window.step1_panel.output_file.get_path() if not water_mask_path and self.work_dir: mask_dir = resolve_subdir(self.work_dir, 'water_mask') if os.path.isdir(mask_dir): dat_files = [f for f in os.listdir(mask_dir) if f.lower().endswith('.dat')] if dat_files: water_mask_path = os.path.join(mask_dir, dat_files[0]).replace('\\', '/') if not water_mask_path and self.work_dir: input_test_dir = os.path.join(self.work_dir, "input-test") if os.path.isdir(input_test_dir): dat_files = [f for f in os.listdir(input_test_dir) if f.lower().endswith('.dat')] for f in dat_files: if 'water_mask_from_shp' in f.lower(): water_mask_path = os.path.join(input_test_dir, f).replace('\\', '/') break if not water_mask_path and dat_files: water_mask_path = os.path.join(input_test_dir, dat_files[0]).replace('\\', '/') if water_mask_path: if not os.path.isabs(water_mask_path): water_mask_path = os.path.join(self.work_dir or '', water_mask_path).replace('\\', '/') self.water_mask_file.set_path(water_mask_path) if self.work_dir: output_path = resolve_subdir(self.work_dir, 'sampling_csv_path') os.makedirs(os.path.dirname(output_path), exist_ok=True) self.output_file.set_path(output_path.replace('\\', '/')) self._check_csv_exists() def _on_run_single_clicked(self): from src.gui.core.event_bus import global_event_bus deglint_img_path = self.deglint_img_file.get_path() if not deglint_img_path: QMessageBox.warning(self, "输入错误", "请选择去耀斑影像文件！") return config = {'step4_sampling': self.get_config()} global_event_bus.publish('RequestRunSingleStep', { 'step_name': 'step4_sampling', 'config': config, }) def _check_csv_exists(self): csv_path = self.output_file.get_path() enabled = bool(csv_path and os.path.isabs(csv_path) and os.path.exists(csv_path)) self.preview_btn.setEnabled(enabled) return enabled def _on_output_changed(self, _text=None): self._check_csv_exists() def _open_sampling_viewer(self): csv_path = self.output_file.get_path() if not csv_path or not os.path.exists(csv_path): QMessageBox.warning(self, "文件不存在", f"采样点 CSV 文件不存在：{csv_path}\n请先运行生成数据。") return dialog = SamplingViewerDialog(csv_path, self) dialog.exec_() self._check_csv_exists()