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

refactor(step10_new_arch): 同步新架构 view + service 到散点 CSV 模式

Browse Source
This commit is contained in:
DXC
2026-06-24 12:44:01 +08:00
parent 08a9a0337d
commit 6a1014afcc
2 changed files with 178 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
src/new/services/step10_service.py
View File

@ -1,36 +1,42 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Step10 后端计算服务（水色指数反演） Step10 后端计算服务（水色指数反演 · 散点 CSV 模式）
==================================== ====================================================
纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只： 纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只：
1. 从 ``config`` 字典读取参数； 1. 从 ``config`` 字典读取参数；
2. 调用 ``WaterIndexProcessor.run_inversion`` 用 ``waterindex.csv`` 中的 2. 调用 ``WaterIndexCsvProcessor.compute_indices_from_csv``
公式直接处理去耀斑 BSQ 影像，输出各水质参数指数的 GeoTIFF； 读取 Step 4 输出的 ``sampling_spectra.csv`` 散点，对每行采样点
3. 返回结果字典 ``{status, output_path, message, mode}``。 套用 ``waterindex.csv`` 中勾选的公式，输出每公式一个 CSV；
3. 返回结果字典 ``{status, output_path, message, mode, ...}``。
调用入口（由 main_view 在后台 QThread 中调用）： 调用入口（由 main_view 在后台 QThread 中调用）：
execute_step10({ execute_step10({
"bsq_path": "D:/deglint_output.bsq", # 去耀斑 BSQ 影像（必填） "sampling_csv_path": "D:/4_sampling/sampling_spectra.csv", # 必填
"deglint_img_path": "D:/deglint_output.bsq", # 同上（兼容旧 panel 字段） "selected_formulas": ["NDCI", "BGA_Am09KBBI"], # 勾选公式；空 → 全部
"hdr_path": "D:/deglint_output.hdr", # ENVI 头文件（可省，自动 .bsq→.hdr 推断） "formula_csv_path": "D:/waterindex.csv", # waterindex.csv 路径
"selected_formulas": ["NDCI", "BGA_Am09KBBI"], # 要处理的公式名列表（空 → 全部） "output_dir": "D:/10_WaterIndex_CSV", # 输出目录；可省
"formula_csv_path": "D:/waterindex.csv", # waterindex.csv 路径（可省，自动探测） "enabled": True,
"water_mask_path": "D:/water_mask.dat", # 水域掩膜路径（可省） "work_dir": "D:/workspace", # 主窗口注入
"nodata_value": -9999.0, # NoData 标记值
"output_dir": "D:/10_WaterIndex_Images", # 输出目录（可省 → work_dir/10_WaterIndex_Images）
"enabled": True,
"work_dir": "D:/workspace", # 工作目录（main_view 注入）
}) })
返回字典字段： 返回字典字段：
* ``status`` : "completed" | "skipped" | "error" * ``status`` : "completed" | "skipped" | "error"
* ``output_path`` : 输出目录路径（失败时为 None） * ``output_path`` : 输出目录路径（失败时为 None）
* ``output_files`` : {公式名: 公式 CSV 路径}（失败时为空 dict）
* ``message`` : 人类可读说明 * ``message`` : 人类可读说明
* ``mode`` : "watercolor_inversion"（便于 UI 提示） * ``mode`` : "watercolor_inversion_csv"（便于 UI 提示）
设计要点
========
- 与 Step 9 (ML 预测) 完全对称的"散点处理模式"：输入 CSV、输出 CSV，
坐标列重命名为 longitude/latitude，公式值以列形式追加。
- 旧"读 BSQ 全图 → 输出 GeoTIFF"模式已废弃（科学上误差大且与 GIS 栅格计算器重复）。
- 兼容调用方可能仍传旧键（bsq_path / hdr_path / deglint_img_path），检测到时
静默忽略并回退到 sampling_spectra.csv 路径解析（避免破坏已有 pipeline 配置）。
""" """
from __future__ import annotations from __future__ import annotations
@ -38,11 +44,40 @@ from __future__ import annotations
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from src.new.services._output_resolver import get_user_output_path, is_user_specified, resolve_output_dir from src.new.services._output_resolver import get_user_output_path
def _resolve_sampling_csv_path(
sampling_csv_path: Optional[str],
work_dir: str,
) -> str:
"""解析采样点 CSV 路径
解析顺序：
1. 显式传入的 ``sampling_csv_path``
2. ``{work_dir}/4_sampling/sampling_spectra.csv``
3. ``{work_dir}/4_sampling/`` 下任意 ``.csv`` (取最新)
"""
if sampling_csv_path and Path(sampling_csv_path).is_file():
return sampling_csv_path
if not work_dir:
return sampling_csv_path or ""
primary = Path(work_dir) / "4_sampling" / "sampling_spectra.csv"
if primary.is_file():
return str(primary).replace("\\", "/")
sample_dir = Path(work_dir) / "4_sampling"
if sample_dir.is_dir():
cands = sorted(sample_dir.glob("*.csv"), key=lambda p: p.stat().st_mtime, reverse=True)
if cands:
return str(cands[0]).replace("\\", "/")
return sampling_csv_path or ""
def _resolve_waterindex_csv(formula_csv_path: Optional[str], work_dir: str) -> str: def _resolve_waterindex_csv(formula_csv_path: Optional[str], work_dir: str) -> str:
"""解析 waterindex.csv 路径（与 WaterIndexProcessor.__init__ 默认逻辑保持一致）""" """解析 waterindex.csv 路径（与 WaterIndexCsvProcessor.__init__ 默认逻辑保持一致）"""
if formula_csv_path and Path(formula_csv_path).is_file(): if formula_csv_path and Path(formula_csv_path).is_file():
return formula_csv_path return formula_csv_path
candidates = [ candidates = [
@ -52,142 +87,118 @@ def _resolve_waterindex_csv(formula_csv_path: Optional[str], work_dir: str) -> s
] ]
for c in candidates: for c in candidates:
if c.is_file(): if c.is_file():
return str(c) return str(c).replace("\\", "/")
return formula_csv_path or "" return formula_csv_path or ""
def _resolve_water_mask_path(water_mask_path: Optional[str], work_dir: str) -> Optional[str]:
"""解析水域掩膜路径（缺省时尝试从 work_dir/1_water_mask 自动扫盘）"""
if water_mask_path and Path(water_mask_path).is_file():
return water_mask_path
if not work_dir:
return None
mask_dir = Path(work_dir) / "1_water_mask"
if not mask_dir.is_dir():
return None
for pat in ("*.tif", "*.TIF", "*.dat", "*.DT"):
cands = sorted(mask_dir.glob(pat))
if cands:
return str(cands[0])
return None
def _resolve_output_dir(config: Dict[str, Any], work_dir: str) -> tuple[Path, str]: def _resolve_output_dir(config: Dict[str, Any], work_dir: str) -> tuple[Path, str]:
"""根据 output_dir / work_dir 计算水色指数反演结果输出目录 """根据 output_dir / work_dir 计算水色指数反演结果输出目录
使用共享解析器强制执行"用户优先"规则——用户指定 output_dir 时直接用其值 使用共享解析器强制执行"用户优先"规则——用户指定 output_dir 时直接用其值
（step10 的 output_dir 本身就是一个目录），否则用 work_dir/10_WaterIndex_Images 默认。 （step10 的 output_dir 本身就是一个目录），否则用
``work_dir/10_WaterIndex_CSV`` 默认。
注意：step10 与其他步骤不同——output_dir 直接表示目录而非文件路径，
所以使用 Path(user_path) 而非 .parent。
""" """
user_path = get_user_output_path(config, "output_dir", "output_path") user_path = get_user_output_path(config, "output_dir", "output_path")
if user_path: if user_path:
return Path(user_path), "user" return Path(user_path), "user"
return Path(work_dir) / "10_WaterIndex_Images", "default" return Path(work_dir) / "10_WaterIndex_CSV", "default"
def execute_step10(config: Dict[str, Any]) -> Dict[str, Any]: def execute_step10(config: Dict[str, Any]) -> Dict[str, Any]:
"""Step 10 后端计算入口——纯函数 """Step 10 后端计算入口——纯函数（散点 CSV 模式）
Args: Args:
config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典 config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典
Returns: Returns:
标准结果字典 ``{status, output_path, message, mode}`` 标准结果字典 ``{status, output_path, output_files, message, mode}``
""" """
# ---------- 入参规整 ---------- # ---------- 入参规整 ----------
bsq_path: str = config.get("bsq_path") or config.get("deglint_img_path") or "" sampling_csv_path: str = (
hdr_path: str = config.get("hdr_path") or "" config.get("sampling_csv_path")
or config.get("spectrum_csv_path") # 兼容旧字段
or ""
)
selected_formulas: List[str] = config.get("selected_formulas") or [] selected_formulas: List[str] = config.get("selected_formulas") or []
formula_csv_path: str = config.get("formula_csv_path") or "" formula_csv_path: str = config.get("formula_csv_path") or ""
water_mask_path: Optional[str] = config.get("water_mask_path")
nodata_value: float = float(config.get("nodata_value", -9999.0))
output_dir: str = config.get("output_dir") or "" output_dir: str = config.get("output_dir") or ""
enabled: bool = bool(config.get("enabled", True)) enabled: bool = bool(config.get("enabled", True))
work_dir: str = config.get("work_dir") or "." work_dir: str = config.get("work_dir") or "."
output_path, _source = _resolve_output_dir(config, work_dir) output_path, _source = _resolve_output_dir(config, work_dir)
mode = "watercolor_inversion" mode = "watercolor_inversion_csv"
# ---------- 提前失败检查 ---------- # ---------- 提前失败检查 ----------
if not enabled: if not enabled:
return { return {
"status": "skipped", "status": "skipped",
"output_path": None, "output_path": None,
"output_files": {},
"message": "用户禁用此步骤（enabled=False）", "message": "用户禁用此步骤（enabled=False）",
"mode": mode, "mode": mode,
} }
if not bsq_path:
return {
"status": "error",
"output_path": None,
"message": "未提供 BSQ 影像路径（bsq_path / deglint_img_path）",
"mode": mode,
}
if not Path(bsq_path).is_file():
return {
"status": "error",
"output_path": None,
"message": f"BSQ 影像不存在: {bsq_path}",
"mode": mode,
}
if not hdr_path:
# 自动探测 .hdr
hdr_path = str(Path(bsq_path).with_suffix(".hdr"))
if not Path(hdr_path).is_file():
hdr_alt = str(Path(bsq_path).with_suffix(".HDR"))
if Path(hdr_alt).is_file():
hdr_path = hdr_alt
else:
hdr_path = ""
if not hdr_path or not Path(hdr_path).is_file(): # 解析采样点 CSV 路径
resolved_sampling_csv = _resolve_sampling_csv_path(sampling_csv_path, work_dir)
if not resolved_sampling_csv:
return { return {
"status": "error", "status": "error",
"output_path": None, "output_path": None,
"message": f"未找到 ENVI 头文件（与 BSQ 同名 .hdr）: {bsq_path}", "output_files": {},
"message": "未提供 sampling_csv_path 且默认位置均找不到 sampling_spectra.csv",
"mode": mode,
}
if not Path(resolved_sampling_csv).is_file():
return {
"status": "error",
"output_path": None,
"output_files": {},
"message": f"采样点 CSV 不存在: {resolved_sampling_csv}",
"mode": mode, "mode": mode,
} }
# ---------- 解析 waterindex.csv ---------- # 解析 waterindex.csv
resolved_formula_csv = _resolve_waterindex_csv(formula_csv_path, work_dir) resolved_formula_csv = _resolve_waterindex_csv(formula_csv_path, work_dir)
if not resolved_formula_csv: if not resolved_formula_csv:
return { return {
"status": "error", "status": "error",
"output_path": None, "output_path": None,
"output_files": {},
"message": "未提供 formula_csv_path 且默认位置均找不到 waterindex.csv", "message": "未提供 formula_csv_path 且默认位置均找不到 waterindex.csv",
"mode": mode, "mode": mode,
} }
if not Path(resolved_formula_csv).is_file():
# ---------- 解析水域掩膜（可选） ---------- return {
resolved_water_mask = _resolve_water_mask_path(water_mask_path, work_dir) "status": "error",
"output_path": None,
"output_files": {},
"message": f"waterindex.csv 不存在: {resolved_formula_csv}",
"mode": mode,
}
# ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ---------- # ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ----------
try: try:
from src.core.algorithms.waterindex_inversion import WaterIndexProcessor from src.core.algorithms.waterindex_inversion import (
WaterIndexCsvProcessor,
)
print(f"[Step10 Service] 水色指数反演: bsq={bsq_path}") print(f"[Step10 Service] 水色指数反演(散点模式): sampling_csv={resolved_sampling_csv}")
print(f"[Step10 Service] hdr={hdr_path}")
print(f"[Step10 Service] formula_csv={resolved_formula_csv}") print(f"[Step10 Service] formula_csv={resolved_formula_csv}")
print(f"[Step10 Service] selected_formulas={selected_formulas or '全部'}") print(f"[Step10 Service] selected_formulas={selected_formulas or '全部'}")
if resolved_water_mask: print(f"[Step10 Service] output_dir={output_path}")
print(f"[Step10 Service] water_mask={resolved_water_mask}")
processor = WaterIndexProcessor(resolved_formula_csv) processor = WaterIndexCsvProcessor(resolved_formula_csv)
results = processor.run_inversion( out_files = processor.compute_indices_from_csv(
deglint_img_path=bsq_path, sampling_csv_path=resolved_sampling_csv,
work_dir=work_dir, output_dir=str(output_path).replace("\\", "/"),
formula_csv_path=resolved_formula_csv,
selected_formulas=selected_formulas or None, selected_formulas=selected_formulas or None,
water_mask_path=resolved_water_mask, progress_callback=None, # 日志由 main_view 统一接管
nodata_value=nodata_value,
callback=None, # 日志由 main_view 统一接管
) )
except FileNotFoundError as e: except FileNotFoundError as e:
return { return {
"status": "error", "status": "error",
"output_path": None, "output_path": None,
"output_files": {},
"message": f"文件不存在: {e}", "message": f"文件不存在: {e}",
"mode": mode, "mode": mode,
} }
@ -195,6 +206,7 @@ def execute_step10(config: Dict[str, Any]) -> Dict[str, Any]:
return { return {
"status": "error", "status": "error",
"output_path": None, "output_path": None,
"output_files": {},
"message": f"参数错误: {e}", "message": f"参数错误: {e}",
"mode": mode, "mode": mode,
} }
@ -202,16 +214,18 @@ def execute_step10(config: Dict[str, Any]) -> Dict[str, Any]:
return { return {
"status": "error", "status": "error",
"output_path": None, "output_path": None,
"output_files": {},
"message": f"{type(e).__name__}: {e}", "message": f"{type(e).__name__}: {e}",
"mode": mode, "mode": mode,
} }
# ---------- 成功路径 ---------- # ---------- 成功路径 ----------
p = Path(output_path) p = Path(output_path)
n_results = len(results) if isinstance(results, dict) else 0 n_results = len(out_files) if isinstance(out_files, dict) else 0
return { return {
"status": "completed", "status": "completed",
"output_path": str(p).replace("\\", "/"), "output_path": str(p).replace("\\", "/"),
"message": f"水色指数反演完成，共生成 {n_results} 个指数 GeoTIFF", "output_files": out_files,
"message": f"水色指数反演完成，共生成 {n_results} 个指数 CSV",
"mode": mode, "mode": mode,
} }

136
src/new/views/step10_view.py
View File

@ -1,20 +1,18 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Step10View —— Step 10（水色指数反演）的端到端模块化 view Step10View —— Step 10（水色指数反演）的端到端模块化 view（散点 CSV 模式）
UI 从 ``src/gui/panels/step10_watercolor_panel.py`` 原样搬迁。 与 ``src/gui/panels/step10_watercolor_panel.py`` 同步重构。
view 层职责 view 层职责
=========== ===========
- 输入影像（BSQ + HDR）、公式选择 ListWidget、输出目录 + 格式 combo、 - 输入采样点 CSV、公式选择 ListWidget、输出目录、进度条 / 进度标签、运行按钮。
进度条 / 进度标签、运行按钮全部保留。 - 删除 BSQ + HDR FileSelectWidget（散点模式不读全图栅格）。
- 删除 ``WaterIndexWorker`` 线程（service 接管后台反演逻辑）； - 删除 ``format_combo``（每个公式一个 CSV，无格式选择）。
进度条 / 标签在 view 层保留 UI 占位，由 service 通过 - 删除 ``WaterIndexWorker`` 线程（service 接管后台计算逻辑）。
``dispatch_execute`` 反馈到主窗口的日志区即可。 - ``_load_formulas`` / 加载 waterindex.csv 的逻辑不在 view 层执行；
- ``_find_waterindex_csv`` / ``_load_formulas`` / ``_load_metadata`` 公式 ListWidget 留空，service 通过 set_config 把 selected_formulas 注入。
不在 view 层执行；公式 ListWidget 留空，service 通过 set_config
把 selected_formulas 注入。
""" """
import os import os
@ -37,50 +35,51 @@ def _resolve_subdir(work_dir: str, subdir_name: str) -> str:
class Step10View(BaseView): class Step10View(BaseView):
"""Step 10: 水色指数反演（高光谱影像直接处理）""" """Step 10: 水色指数反演（散点 CSV 模式）
输入：Step 4 输出的 sampling_spectra.csv（散点 + 全波段光谱）
处理：逐行套用 waterindex.csv 公式
输出：每公式一个 CSV（列：longitude, latitude, 公式值）
"""
def init_ui(self): def init_ui(self):
layout = QVBoxLayout() layout = QVBoxLayout()
# ---- 标题 ---- # ---- 标题 ----
title = QLabel("步骤10：水色指数反演（高光谱影像直接处理）") title = QLabel("步骤10：水色指数反演（散点 CSV 模式）")
title.setFont(QFont("Arial", 12, QFont.Bold)) title.setFont(QFont("Arial", 12, QFont.Bold))
layout.addWidget(title) layout.addWidget(title)
# ---- 说明 ---- # ---- 说明 ----
hint = QLabel( hint = QLabel(
"将 waterindex.csv 中的公式直接应用于去耀斑高光谱影像（BSQ），" "读取 Step 4 生成的 sampling_spectra.csv 散点光谱，"
"输出各水质参数指数的 GeoTIFF 栅格图像。" "对每个采样点逐行套用 waterindex.csv 中勾选的公式，"
"指数图可直接用于水质专题图生成。" "输出每公式一个 CSV（列：longitude, latitude, 公式值）。"
"结果可被 Step 11 直接以 ContentMapper 模式消费。"
) )
hint.setWordWrap(True) hint.setWordWrap(True)
hint.setStyleSheet(f"color: {ModernStylesheet.COLORS.get('text_secondary', '#666')};") hint.setStyleSheet(f"color: {ModernStylesheet.COLORS.get('text_secondary', '#666')};")
layout.addWidget(hint) layout.addWidget(hint)
# ---- 输入影像选择 ---- # ---- 输入采样点数据 ----
input_group = QGroupBox("输入影像") input_group = QGroupBox("输入采样点数据")
input_layout = QFormLayout() input_layout = QFormLayout()
self.bsq_file = FileSelectWidget( self.sampling_csv_file = FileSelectWidget(
"BSQ 影像:", "采样点 CSV:",
"BSQ Files (*.bsq);;DAT Files (*.dat);;All Files (*.*)", "CSV Files (*.csv);;All Files (*.*)",
) )
self.bsq_file.line_edit.setPlaceholderText("选择去耀斑处理后的 BSQ 影像") self.sampling_csv_file.line_edit.setPlaceholderText(
input_layout.addRow("BSQ 影像:", self.bsq_file) "选择 Step 4 输出的 sampling_spectra.csv"
self.hdr_file = FileSelectWidget(
"ENVI 头文件:",
"HDR Files (*.hdr);;All Files (*.*)",
) )
self.hdr_file.line_edit.setPlaceholderText("自动关联同路径 .hdr 文件") input_layout.addRow("采样点 CSV:", self.sampling_csv_file)
input_layout.addRow("HDR 文件:", self.hdr_file)
self.meta_label = QLabel("未加载影像") self.meta_label = QLabel("未加载采样点数据")
self.meta_label.setStyleSheet( self.meta_label.setStyleSheet(
"background: #f0f0f0; padding: 4px 8px; border-radius: 4px; " "background: #f0f0f0; padding: 4px 8px; border-radius: 4px; "
"font-size: 12px; color: #333;" "font-size: 12px; color: #333;"
) )
input_layout.addRow("影像信息:", self.meta_label) input_layout.addRow("数据信息:", self.meta_label)
input_group.setLayout(input_layout) input_group.setLayout(input_layout)
layout.addWidget(input_group) layout.addWidget(input_group)
@ -125,14 +124,11 @@ class Step10View(BaseView):
"输出目录:", "输出目录:",
"Directories", "Directories",
) )
self.output_dir.line_edit.setPlaceholderText("留空 → 工作目录/10_WaterIndex_Images") self.output_dir.line_edit.setPlaceholderText(
"留空 → 工作目录/10_WaterIndex_CSV"
)
output_layout.addRow("输出目录:", self.output_dir) output_layout.addRow("输出目录:", self.output_dir)
self.format_combo = QComboBox()
self.format_combo.addItems(["GTiff (GeoTIFF)", "ENVI", "PCI"])
self.format_combo.setCurrentIndex(0)
output_layout.addRow("输出格式:", self.format_combo)
output_group.setLayout(output_layout) output_group.setLayout(output_layout)
layout.addWidget(output_group) layout.addWidget(output_group)
@ -178,7 +174,7 @@ class Step10View(BaseView):
# BaseView 契约 # BaseView 契约
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def get_config(self) -> dict: def get_config(self) -> dict:
bsq_path = self.bsq_file.get_path() sampling = self.sampling_csv_file.get_path().strip()
selected = [] selected = []
for i in range(self.formula_list.count()): for i in range(self.formula_list.count()):
item = self.formula_list.item(i) item = self.formula_list.item(i)
@ -186,26 +182,19 @@ class Step10View(BaseView):
name = item.data(Qt.UserRole) name = item.data(Qt.UserRole)
if name: if name:
selected.append(name) selected.append(name)
config = { config: dict = {
"bsq_path": bsq_path, "sampling_csv_path": sampling,
"deglint_img_path": bsq_path,
"output_format": self.format_combo.currentText().split()[0],
"selected_formulas": selected, "selected_formulas": selected,
"enabled": self.enable_checkbox.isChecked(), "enabled": self.enable_checkbox.isChecked(),
} }
hdr_path = self.hdr_file.get_path() output_dir = self.output_dir.get_path().strip()
if hdr_path:
config["hdr_path"] = hdr_path
output_dir = self.output_dir.get_path()
if output_dir: if output_dir:
config["output_dir"] = output_dir config["output_dir"] = output_dir
return config return config
def set_config(self, config: dict): def set_config(self, config: dict):
if config.get("bsq_path"): if config.get("sampling_csv_path"):
self.bsq_file.set_path(config["bsq_path"]) self.sampling_csv_file.set_path(config["sampling_csv_path"])
if config.get("hdr_path"):
self.hdr_file.set_path(config["hdr_path"])
if config.get("output_dir"): if config.get("output_dir"):
self.output_dir.set_path(config["output_dir"]) self.output_dir.set_path(config["output_dir"])
if "selected_formulas" in config: if "selected_formulas" in config:
@ -223,25 +212,44 @@ class Step10View(BaseView):
super().update_work_directory(work_dir) super().update_work_directory(work_dir)
if not work_dir: if not work_dir:
return return
out_dir = _resolve_subdir(work_dir, "watercolor")
# 1) 自动填采样点 CSV（从 step4 拉取）
sampling_path = self._find_step4_sampling_csv(work_dir)
if sampling_path and not self.sampling_csv_file.get_path():
self.sampling_csv_file.set_path(sampling_path)
# 2) 自动填输出目录
out_dir = os.path.join(work_dir, "10_WaterIndex_CSV").replace("\\", "/")
os.makedirs(out_dir, exist_ok=True) os.makedirs(out_dir, exist_ok=True)
if not self.output_dir.get_path(): if not self.output_dir.get_path():
self.output_dir.set_path(out_dir) self.output_dir.set_path(out_dir)
# 自动填 BSQ（去耀斑输出）
deglint_dir = _resolve_subdir(work_dir, "deglint") def _find_step4_sampling_csv(self, work_dir: str) -> str:
if os.path.isdir(deglint_dir): """从 step4 panel / pipeline.step_outputs / 4_sampling 目录自动找采样点 CSV"""
# 1) 优先：从主窗口懒加载面板读
mw = self.window()
factory = getattr(mw, "_panel_factory", None) if mw else None
if factory:
step4_panel = factory.get_panel("step4_sampling")
if step4_panel and hasattr(step4_panel, "output_file"):
p = step4_panel.output_file.get_path().strip()
if p and os.path.isfile(p):
return p
# 2) 兜底：扫 4_sampling/sampling_spectra.csv
candidate = os.path.join(work_dir, "4_sampling", "sampling_spectra.csv")
if os.path.isfile(candidate):
return candidate.replace("\\", "/")
# 3) 终极兜底：扫 4_sampling 下任意 .csv
sample_dir = os.path.join(work_dir, "4_sampling")
if os.path.isdir(sample_dir):
import glob import glob
candidates = ( cands = glob.glob(os.path.join(sample_dir, "*.csv"))
glob.glob(os.path.join(deglint_dir, "*.bsq")) if cands:
+ glob.glob(os.path.join(deglint_dir, "*.dat")) cands.sort(key=os.path.getmtime, reverse=True)
) return cands[0].replace("\\", "/")
if candidates and not self.bsq_file.get_path(): return ""
candidates.sort(key=os.path.getmtime, reverse=True)
bsq_path = candidates[0]
self.bsq_file.set_path(bsq_path)
hdr_path = os.path.splitext(bsq_path)[0] + ".hdr"
if os.path.exists(hdr_path):
self.hdr_file.set_path(hdr_path)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 执行入口 # 执行入口