#!/usr/bin/env python # -*- coding: utf-8 -*- """ Step4 面板 - 采样点布设（内嵌交互式光谱探针视图） 2026-06-30 重构： - 左右分栏布局 (QSplitter)：左侧控制区 + 右侧嵌入式 Matplotlib 视图 - 1×2 子图：ax1 散点图 + ax2 光谱曲线 - Hover 悬停显示坐标提示，Click 点击绘制该点光谱曲线 - NavigationToolbar2QT 工具栏自带保存/缩放/平移 """ import os import sys from pathlib import Path import numpy as np import pandas as pd _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, QSpinBox, QCheckBox, QMessageBox, QLabel, QFrame, QSplitter, QSizePolicy, ) from matplotlib.backends.backend_qt5agg import ( FigureCanvasQTAgg as FigureCanvas, NavigationToolbar2QT as NavigationToolbar, ) from matplotlib.figure import Figure from src.gui.components.custom_widgets import FileSelectWidget from src.gui.dialogs import SamplingViewerDialog from src.gui.styles import ModernStylesheet class Step4SamplingPanel(QWidget): def __init__(self, parent=None): super().__init__(parent) # 交互状态 self._df = None self._x_col = None self._y_col = None self._band_cols = [] self._scatter = None self._highlight_idx = None self._annot = None self._cid_hover = None self._cid_click = None self._last_render_path = None self.init_ui() # ═══════════════════════════════════════════════════════════════ # UI 构建 # ═══════════════════════════════════════════════════════════════ def init_ui(self): self.setStyleSheet(ModernStylesheet.get_main_stylesheet()) # ── 顶层：水平分栏 (QSplitter) ── splitter = QSplitter(Qt.Horizontal) splitter.setChildrenCollapsible(False) # ═══════════════════════════════════════════════ # 左侧：控制区（原封保留原有三张卡片） # ═══════════════════════════════════════════════ left_widget = QWidget() left_layout = QVBoxLayout() left_layout.setContentsMargins(24, 24, 12, 24) left_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) left_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) left_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() action_layout.addStretch() self.refresh_btn = QPushButton("🔄 刷新视图") self.refresh_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('normal')) self.refresh_btn.setEnabled(False) self.refresh_btn.setMinimumWidth(140) self.refresh_btn.setToolTip("重新加载 CSV 并渲染采样点散点图") self.refresh_btn.clicked.connect(self._on_refresh_clicked) 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.refresh_btn) action_layout.addWidget(self.run_btn) output_layout.addLayout(action_layout) output_group.setLayout(output_layout) left_layout.addWidget(output_group) left_layout.addStretch() left_widget.setLayout(left_layout) left_widget.setMinimumWidth(360) # ═══════════════════════════════════════════════ # 右侧：可视化区（嵌入式 Matplotlib 视图） # ═══════════════════════════════════════════════ right_widget = QWidget() right_layout = QVBoxLayout() right_layout.setContentsMargins(12, 24, 24, 24) right_layout.setSpacing(0) viz_group = QGroupBox("📊 采样点交互式探索") viz_layout = QVBoxLayout() viz_layout.setContentsMargins(8, 20, 8, 8) viz_layout.setSpacing(0) # Matplotlib 画布（1×2 子图：散点图 + 光谱曲线） self._fig = Figure(figsize=(9, 5)) self._ax_scatter = self._fig.add_subplot(121) self._ax_spectrum = self._fig.add_subplot(122) self._canvas = FigureCanvas(self._fig) self._canvas.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) # 设置子图初始状态 self._ax_scatter.set_title("采样点分布", fontsize=11, fontweight='bold') self._ax_scatter.set_xlabel("X 坐标") self._ax_scatter.set_ylabel("Y 坐标") self._ax_scatter.text(0.5, 0.5, "等待采样数据生成...\n\n请先配置参数并运行步骤\n或选择已有的 CSV 后刷新", ha='center', va='center', transform=self._ax_scatter.transAxes, fontsize=12, color='#888888') self._ax_scatter.grid(False) self._ax_spectrum.set_title("光谱曲线", fontsize=11, fontweight='bold') self._ax_spectrum.set_xlabel("波长 (nm)") self._ax_spectrum.set_ylabel("反射率") self._ax_spectrum.text(0.5, 0.5, "点击左侧散点\n查看光谱曲线", ha='center', va='center', transform=self._ax_spectrum.transAxes, fontsize=12, color='#888888') self._ax_spectrum.grid(False) self._fig.tight_layout(pad=2.0) # 工具栏（自带保存/缩放/平移） self._toolbar = NavigationToolbar(self._canvas, self) viz_layout.addWidget(self._toolbar) viz_layout.addWidget(self._canvas) viz_group.setLayout(viz_layout) right_layout.addWidget(viz_group) right_widget.setLayout(right_layout) # ── 组装分栏 ── splitter.addWidget(left_widget) splitter.addWidget(right_widget) splitter.setSizes([420, 680]) # 初始比例 ≈ 38:62 splitter.setStretchFactor(0, 1) splitter.setStretchFactor(1, 2) top_layout = QHBoxLayout() top_layout.setContentsMargins(0, 0, 0, 0) top_layout.addWidget(splitter) self.setLayout(top_layout) # ── 事件绑定 ── self._cid_hover = self._canvas.mpl_connect('motion_notify_event', self._on_hover) self._cid_click = self._canvas.mpl_connect('button_press_event', self._on_click) # ── 定时器：降低频率，仅用于自动发现新生成的 CSV ── self._status_timer = QTimer(self) self._status_timer.timeout.connect(self._check_csv_and_auto_render) self._status_timer.start(5000) 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('\\', '/') if os.path.exists(deglint_path): 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('\\', '/') if os.path.exists(water_mask_path): self.water_mask_file.set_path(water_mask_path) if self.work_dir and not self.output_file.get_path(): output_path = resolve_subdir(self.work_dir, 'sampling_csv_path') self.output_file.set_path(output_path.replace('\\', '/')) self._check_csv_exists() # 若 CSV 已存在，尝试自动渲染 csv_path = self.output_file.get_path() if csv_path and os.path.isfile(csv_path): self._render_inline_plot() # ═══════════════════════════════════════════════════════════════ # 执行 # ═══════════════════════════════════════════════════════════════ 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, }) # ═══════════════════════════════════════════════════════════════ # CSV 状态检测 # ═══════════════════════════════════════════════════════════════ 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.refresh_btn.setEnabled(enabled) return enabled def _on_output_changed(self, _text=None): self._check_csv_exists() def _check_csv_and_auto_render(self): """定时器回调：检测到新 CSV 出现时自动渲染一次。""" csv_path = self.output_file.get_path() if csv_path and os.path.isfile(csv_path): if csv_path != self._last_render_path: self.refresh_btn.setEnabled(True) self._render_inline_plot() def _on_refresh_clicked(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 self._render_inline_plot() # ═══════════════════════════════════════════════════════════════ # 核心渲染 # ═══════════════════════════════════════════════════════════════ def _detect_coordinate_columns(self, df: pd.DataFrame): """检测坐标列，返回 (x_col, y_col) 或 (None, None)。 优先级：pixel_x/pixel_y → longitude/latitude → lon/lat → X/Y → UTM_X/UTM_Y → 任何含 'x'/'y' 关键字的列 """ cols_lower = {c.lower(): c for c in df.columns} priority_pairs = [ ('pixel_x', 'pixel_y'), ('longitude', 'latitude'), ('lon', 'lat'), ('x', 'y'), ('utm_x', 'utm_y'), ] for x_key, y_key in priority_pairs: if x_key in cols_lower and y_key in cols_lower: return cols_lower[x_key], cols_lower[y_key] # 最后尝试：找名字中含 x / y 的数值列 x_candidates = [c for c in df.columns if 'x' in c.lower() and pd.api.types.is_numeric_dtype(df[c])] y_candidates = [c for c in df.columns if 'y' in c.lower() and pd.api.types.is_numeric_dtype(df[c])] if x_candidates and y_candidates: return x_candidates[0], y_candidates[0] return None, None def _detect_band_columns(self, df: pd.DataFrame): """检测光谱波段列（纯数字列名，值域在 200–3000 nm 之间）。 优先使用列名可解析为 float 且在波长范围内的列； 否则回退到位置索引（跳过坐标列和已知元数据列）。 """ band_cols = [] for col in df.columns: try: val = float(str(col).strip()) if 200.0 <= val <= 3000.0: band_cols.append(col) except (ValueError, TypeError): continue if band_cols: # 按波长数值排序 band_cols.sort(key=lambda c: float(str(c).strip())) return band_cols # 回退：跳过坐标列和已知元数据列，取数值列 skip_keywords = ('x', 'y', 'lon', 'lat', 'utm', 'id', 'sample', 'index', 'pixel') numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist() if self._x_col: numeric_cols = [c for c in numeric_cols if c != self._x_col] if self._y_col: numeric_cols = [c for c in numeric_cols if c != self._y_col] return [c for c in numeric_cols if not any(k in c.lower() for k in skip_keywords)] def _render_inline_plot(self): """读取 CSV → 检测坐标/波段列 → 绘制 1×2 子图。""" csv_path = self.output_file.get_path() if not csv_path or not os.path.isfile(csv_path): self._show_empty_state("等待采样数据生成...\n\n请先配置参数并运行步骤") return # 读取数据 try: df = pd.read_csv(csv_path) except Exception as e: self._show_empty_state(f"读取 CSV 失败:\n{str(e)[:200]}") return if df.empty: self._show_empty_state("CSV 文件为空") return # 检测坐标列 x_col, y_col = self._detect_coordinate_columns(df) if x_col is None or y_col is None: self._show_empty_state( "缺少坐标列\n\n" f"可用列: {', '.join(str(c) for c in df.columns[:15])}\n" "期望: pixel_x/pixel_y, longitude/latitude, X/Y 等" ) return # 检测光谱列 band_cols = self._detect_band_columns(df) if not band_cols: # 仅显示散点图，光谱子图留空 pass # 缓存 self._df = df self._x_col = x_col self._y_col = y_col self._band_cols = band_cols self._highlight_idx = None self._last_render_path = csv_path # 清除 Annotation if self._annot is not None: try: self._annot.remove() except Exception: pass self._annot = None # ── 绘制 ax1：散点图 ── self._ax_scatter.clear() x = df[x_col].values y = df[y_col].values self._scatter = self._ax_scatter.scatter( x, y, c='#0078D7', alpha=0.7, edgecolors='white', linewidth=0.5, s=40, picker=True, zorder=3 ) self._ax_scatter.set_xlabel(str(x_col), fontsize=10) self._ax_scatter.set_ylabel(str(y_col), fontsize=10) self._ax_scatter.set_title(f"采样点分布 (共 {len(df)} 个点)", fontsize=11, fontweight='bold') self._ax_scatter.grid(True, alpha=0.3, linestyle='--') self._ax_scatter.set_facecolor('#F8F9FA') # ── 绘制 ax2：光谱曲线（初始状态）── self._ax_spectrum.clear() if band_cols: self._ax_spectrum.set_title("光谱曲线（点击左侧散点查看）", fontsize=11, fontweight='bold') self._ax_spectrum.set_xlabel("波长 (nm)", fontsize=10) self._ax_spectrum.set_ylabel("反射率", fontsize=10) self._ax_spectrum.grid(True, alpha=0.3, linestyle='--') self._ax_spectrum.text(0.5, 0.5, "点击左侧散点\n查看光谱曲线", ha='center', va='center', transform=self._ax_spectrum.transAxes, fontsize=12, color='#999999') else: self._ax_spectrum.text(0.5, 0.5, "缺少光谱数据列", ha='center', va='center', transform=self._ax_spectrum.transAxes, fontsize=12, color='#999999') self._fig.tight_layout(pad=2.0) self._canvas.draw() def _show_empty_state(self, message: str): """在右侧画布显示提示信息。""" self._ax_scatter.clear() self._ax_scatter.set_title("采样点分布", fontsize=11, fontweight='bold') self._ax_scatter.text(0.5, 0.5, message, ha='center', va='center', transform=self._ax_scatter.transAxes, fontsize=12, color='#888888') self._ax_scatter.grid(False) self._ax_spectrum.clear() self._ax_spectrum.set_title("光谱曲线", fontsize=11, fontweight='bold') self._ax_spectrum.text(0.5, 0.5, "等待数据...", ha='center', va='center', transform=self._ax_spectrum.transAxes, fontsize=12, color='#888888') self._ax_spectrum.grid(False) self._fig.tight_layout(pad=2.0) self._canvas.draw() self._df = None self._last_render_path = None # ═══════════════════════════════════════════════════════════════ # 交互事件 (Hover & Click) # ═══════════════════════════════════════════════════════════════ def _on_hover(self, event): """悬停：在散点旁显示坐标 + ID 提示框。""" if event.inaxes != self._ax_scatter: # 鼠标离开 ax1 → 隐藏 annotation if self._annot is not None: try: self._annot.set_visible(False) self._canvas.draw_idle() except Exception: pass return if self._df is None or self._scatter is None: return # 检测是否悬停在散点上 contains, info = self._scatter.contains(event) if not contains or info is None or 'ind' not in info or len(info['ind']) == 0: if self._annot is not None: try: self._annot.set_visible(False) self._canvas.draw_idle() except Exception: pass return idx = info['ind'][0] row = self._df.iloc[idx] x_val = row[self._x_col] y_val = row[self._y_col] # 创建或更新 Annotation text = f"#{idx}\n({x_val:.4f}, {y_val:.4f})" if self._annot is None: self._annot = self._ax_scatter.annotate( text, xy=(x_val, y_val), xytext=(12, 12), textcoords='offset points', bbox=dict(boxstyle='round,pad=0.4', facecolor='#FFFFFF', edgecolor='#0078D7', alpha=0.9), fontsize=9, zorder=10, ) else: self._annot.xy = (x_val, y_val) self._annot.set_text(text) self._annot.set_visible(True) self._canvas.draw_idle() def _on_click(self, event): """点击：高亮散点 + 绘制光谱曲线。""" if event.inaxes != self._ax_scatter: return if self._df is None or self._scatter is None: return # 检测是否点击在散点上 contains, info = self._scatter.contains(event) if not contains or info is None or 'ind' not in info or len(info['ind']) == 0: return idx = info['ind'][0] row = self._df.iloc[idx] self._highlight_idx = idx # ── 高亮选中点 ── x = self._df[self._x_col].values y = self._df[self._y_col].values colors = ['#0078D7'] * len(self._df) sizes = [40] * len(self._df) colors[idx] = '#E74C3C' sizes[idx] = 80 self._ax_scatter.clear() self._scatter = self._ax_scatter.scatter( x, y, c=colors, s=sizes, alpha=0.7, edgecolors='white', linewidth=0.5, picker=True, zorder=3 ) # 将选中点提升到顶层 self._ax_scatter.scatter( [x[idx]], [y[idx]], c='#E74C3C', s=100, alpha=0.9, edgecolors='white', linewidth=1.5, zorder=5 ) self._ax_scatter.set_xlabel(str(self._x_col), fontsize=10) self._ax_scatter.set_ylabel(str(self._y_col), fontsize=10) self._ax_scatter.set_title(f"采样点分布 (共 {len(self._df)} 个点)", fontsize=11, fontweight='bold') self._ax_scatter.grid(True, alpha=0.3, linestyle='--') self._ax_scatter.set_facecolor('#F8F9FA') # ── 绘制光谱曲线 ── self._ax_spectrum.clear() if self._band_cols: wavelengths = [] reflectance = [] for col in self._band_cols: try: wl = float(str(col).strip()) val = row[col] if pd.notna(val): wavelengths.append(wl) reflectance.append(float(val)) except (ValueError, TypeError): continue if wavelengths: # 按波长排序 pairs = sorted(zip(wavelengths, reflectance), key=lambda p: p[0]) wavelengths, reflectance = zip(*pairs) if pairs else ([], []) self._ax_spectrum.plot( wavelengths, reflectance, color='#0078D7', lw=1.5, marker='.', markersize=3, alpha=0.8 ) self._ax_spectrum.fill_between(wavelengths, reflectance, alpha=0.1, color='#0078D7') self._ax_spectrum.set_xlabel("波长 (nm)", fontsize=10) self._ax_spectrum.set_ylabel("反射率", fontsize=10) self._ax_spectrum.set_title(f"样本 #{idx} 的光谱曲线 ({len(wavelengths)} 个波段)", fontsize=11, fontweight='bold') self._ax_spectrum.grid(True, alpha=0.3, linestyle='--') else: self._ax_spectrum.text(0.5, 0.5, "该样本无有效光谱数据", ha='center', va='center', transform=self._ax_spectrum.transAxes, fontsize=12, color='#999999') else: self._ax_spectrum.text(0.5, 0.5, "缺少光谱波段列\n无法绘制光谱", ha='center', va='center', transform=self._ax_spectrum.transAxes, fontsize=12, color='#999999') self._fig.tight_layout(pad=2.0) self._canvas.draw() # ═══════════════════════════════════════════════════════════════ # 旧版弹窗查看器（保留，供外部调用） # ═══════════════════════════════════════════════════════════════ def _open_sampling_viewer(self): """打开独立的 SamplingViewerDialog 弹窗（保留兼容）。""" 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()