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

services/step10-13：终极决战！打通空间插值、可视化出图与报告生成的最后四步独立服务

Browse Source
This commit is contained in:
DXC
2026-06-17 09:57:13 +08:00
parent 6fc0394fe2
commit 48668c9e74
6 changed files with 1076 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/new/main_view.py
View File

@ -138,32 +138,32 @@ ROUTES = [
"name": "10. 水色指数反演",
"view_module": "src.new.views.step10_view",
"view_class": "Step10View",
"service_module": "src.new.services.placeholder_service",
"service_func": "execute_placeholder",
"service_module": "src.new.services.step10_service",
"service_func": "execute_step10",
},
{
"id": "step11",
"name": "11. 专题图生成",
"view_module": "src.new.views.step11_view",
"view_class": "Step11View",
"service_module": "src.new.services.placeholder_service",
"service_func": "execute_placeholder",
"service_module": "src.new.services.step11_service",
"service_func": "execute_step11",
},
{
"id": "step12",
"name": "12. 可视化",
"view_module": "src.new.views.step12_view",
"view_class": "Step12View",
"service_module": "src.new.services.placeholder_service",
"service_func": "execute_placeholder",
"service_module": "src.new.services.step12_service",
"service_func": "execute_step12",
},
{
"id": "step13",
"name": "13. 报告生成",
"view_module": "src.new.views.step13_view",
"view_class": "Step13View",
"service_module": "src.new.services.placeholder_service",
"service_func": "execute_placeholder",
"service_module": "src.new.services.step13_service",
"service_func": "execute_step13",
},
]
@ -233,7 +233,7 @@ class MainView(QMainWindow):
self._build_routes()
self._log("[Boot] MainView 初始化完成")
self._log(f"[Boot] 已注册 {len(ROUTES)} 条路由（13 个 view 全部真实，前 9 个步骤 service 已迁移）")
self._log(f"[Boot] 已注册 {len(ROUTES)} 条路由（13 个 view 和 service 已全部打通真实链路！）")
# ------------------------------------------------------------------
# UI 布局：左侧导航 + 右侧 stacked + 底部日志

207
src/new/services/step10_service.py Normal file
View File

@ -0,0 +1,207 @@
# -*- coding: utf-8 -*-
"""
Step10 后端计算服务（水色指数反演）
====================================
纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只：
1. 从 ``config`` 字典读取参数；
2. 调用 ``WaterIndexProcessor.run_inversion`` 用 ``waterindex.csv`` 中的
公式直接处理去耀斑 BSQ 影像，输出各水质参数指数的 GeoTIFF；
3. 返回结果字典 ``{status, output_path, message, mode}``。
调用入口（由 main_view 在后台 QThread 中调用）：
execute_step10({
"bsq_path": "D:/deglint_output.bsq", # 去耀斑 BSQ 影像（必填）
"deglint_img_path": "D:/deglint_output.bsq", # 同上（兼容旧 panel 字段）
"hdr_path": "D:/deglint_output.hdr", # ENVI 头文件（可省，自动 .bsq→.hdr 推断）
"selected_formulas": ["NDCI", "BGA_Am09KBBI"], # 要处理的公式名列表（空 → 全部）
"formula_csv_path": "D:/waterindex.csv", # waterindex.csv 路径（可省，自动探测）
"water_mask_path": "D:/water_mask.dat", # 水域掩膜路径（可省）
"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"
* ``output_path`` : 输出目录路径（失败时为 None）
* ``message`` : 人类可读说明
* ``mode`` : "watercolor_inversion"（便于 UI 提示）
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, Optional
def _resolve_waterindex_csv(formula_csv_path: Optional[str], work_dir: str) -> str:
"""解析 waterindex.csv 路径（与 WaterIndexProcessor.__init__ 默认逻辑保持一致）"""
if formula_csv_path and Path(formula_csv_path).is_file():
return formula_csv_path
candidates = [
Path(work_dir) / "waterindex.csv",
Path(work_dir) / "model" / "waterindex.csv",
Path("src/gui/model/waterindex.csv"),
]
for c in candidates:
if c.is_file():
return str(c)
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(output_dir: Optional[str], work_dir: str) -> Path:
"""根据 output_dir / work_dir 计算水色指数反演结果输出目录"""
if output_dir:
return Path(output_dir)
return Path(work_dir) / "10_WaterIndex_Images"
def execute_step10(config: Dict[str, Any]) -> Dict[str, Any]:
"""Step 10 后端计算入口——纯函数
Args:
config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典
Returns:
标准结果字典 ``{status, output_path, message, mode}``
"""
# ---------- 入参规整 ----------
bsq_path: str = config.get("bsq_path") or config.get("deglint_img_path") or ""
hdr_path: str = config.get("hdr_path") or ""
selected_formulas: List[str] = config.get("selected_formulas") 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 ""
enabled: bool = bool(config.get("enabled", True))
work_dir: str = config.get("work_dir") or "."
output_path = _resolve_output_dir(output_dir, work_dir)
mode = "watercolor_inversion"
# ---------- 提前失败检查 ----------
if not enabled:
return {
"status": "skipped",
"output_path": None,
"message": "用户禁用此步骤（enabled=False）",
"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():
return {
"status": "error",
"output_path": None,
"message": f"未找到 ENVI 头文件（与 BSQ 同名 .hdr）: {bsq_path}",
"mode": mode,
}
# ---------- 解析 waterindex.csv ----------
resolved_formula_csv = _resolve_waterindex_csv(formula_csv_path, work_dir)
if not resolved_formula_csv:
return {
"status": "error",
"output_path": None,
"message": "未提供 formula_csv_path 且默认位置均找不到 waterindex.csv",
"mode": mode,
}
# ---------- 解析水域掩膜（可选） ----------
resolved_water_mask = _resolve_water_mask_path(water_mask_path, work_dir)
# ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ----------
try:
from src.core.algorithms.waterindex_inversion import WaterIndexProcessor
print(f"[Step10 Service] 水色指数反演: bsq={bsq_path}")
print(f"[Step10 Service] hdr={hdr_path}")
print(f"[Step10 Service] formula_csv={resolved_formula_csv}")
print(f"[Step10 Service] selected_formulas={selected_formulas or '全部'}")
if resolved_water_mask:
print(f"[Step10 Service] water_mask={resolved_water_mask}")
processor = WaterIndexProcessor(resolved_formula_csv)
results = processor.run_inversion(
deglint_img_path=bsq_path,
work_dir=work_dir,
formula_csv_path=resolved_formula_csv,
selected_formulas=selected_formulas or None,
water_mask_path=resolved_water_mask,
nodata_value=nodata_value,
callback=None, # 日志由 main_view 统一接管
)
except FileNotFoundError as e:
return {
"status": "error",
"output_path": None,
"message": f"文件不存在: {e}",
"mode": mode,
}
except ValueError as e:
return {
"status": "error",
"output_path": None,
"message": f"参数错误: {e}",
"mode": mode,
}
except Exception as e: # noqa: BLE001 —— service 层兜底捕获所有
return {
"status": "error",
"output_path": None,
"message": f"{type(e).__name__}: {e}",
"mode": mode,
}
# ---------- 成功路径 ----------
p = Path(output_path)
n_results = len(results) if isinstance(results, dict) else 0
return {
"status": "completed",
"output_path": str(p).replace("\\", "/"),
"message": f"水色指数反演完成，共生成 {n_results} 个指数 GeoTIFF",
"mode": mode,
}

287
src/new/services/step11_service.py Normal file
View File

@ -0,0 +1,287 @@
# -*- coding: utf-8 -*-
"""
Step11 后端计算服务（专题图生成 / 克里金插值）
==========================================
纯计算函数——绝对不引用 PyQt、绝对不引用 main_view。它只：
1. 从 ``config`` 字典读取参数；
2. 根据 ``render_mode`` 分发到两条独立链路：
- **CSV 插值模式** —— 调用 ``batch_kriging_interpolation`` 把离散的
预测点 CSV 批量插值成栅格化专题图 GeoTIFF；
- **GeoTIFF 栅格模式** —— 调用 ``ContentMapper.visualize_raster`` 把
水色指数 GeoTIFF 直接渲染成 PNG 专题图（不依赖克里金）；
3. 返回结果字典 ``{status, output_path, message, mode}``。
调用入口（由 main_view 在后台 QThread 中调用）：
execute_step11({
"render_mode": "CSV 插值模式" | "GeoTIFF 栅格模式",
"step10_batch_mode": "single" | "folder",
"prediction_csv_path": "D:/9_ML_Prediction/Chl.csv", # 单文件（CSV 模式）
"prediction_csv_dir": "D:/9_ML_Prediction", # 批量目录
"geotiff_path": "D:/10_WaterIndex_Images/Chl_NDCI.tif", # 单 GeoTIFF
"geotiff_dir": "D:/10_WaterIndex_Images", # 批量 GeoTIFF
"boundary_shp_path": "D:/boundary.shp", # 边界 shp（可选）
"resolution": 30.0, # 空间分辨率（米）
"input_crs": "EPSG:32651",
"output_crs": "EPSG:4326",
"output_dir": "D:/11_Thematic_Map", # 输出目录
"enabled": True,
"work_dir": "D:/workspace", # 工作目录
})
返回字典字段：
* ``status`` : "completed" | "skipped" | "error"
* ``output_path`` : 输出目录或文件路径（失败时为 None）
* ``message`` : 人类可读说明
* ``mode`` : "make_thematic_map"（便于 UI 提示）
设计取舍
--------
- CSV 模式统一用 ``batch_kriging_interpolation``（支持单/批、含参考影像投影读取）；
- GeoTIFF 模式统一用 ``ContentMapper.visualize_raster``（matplotlib Agg 无头后端，
service 内 set_backend 保证线程内可重入）。
- view 层只负责收集配置；service 端负责 matplotlib backend 切换与恢复。
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, Optional
def _resolve_output_dir(output_dir: Optional[str], work_dir: str) -> Path:
"""根据 output_dir / work_dir 计算专题图输出目录"""
if output_dir:
return Path(output_dir)
return Path(work_dir) / "11_Thematic_Map"
def _resolve_csv_paths(config: Dict[str, Any], work_dir: str) -> List[Path]:
"""解析预测 CSV 路径列表（支持单文件 / 文件夹批量 / 递归）"""
paths: List[Path] = []
single = config.get("prediction_csv_path")
folder = config.get("prediction_csv_dir")
recursive = bool(config.get("recursive_csv_scan", False))
if single:
p = Path(single)
if p.is_file():
paths.append(p)
if folder:
d = Path(folder)
if d.is_dir():
glob_method = d.rglob if recursive else d.glob
paths.extend(sorted(p for p in glob_method("*.csv") if p.is_file()))
if not paths and work_dir:
# 兜底：自动扫 9_ML_Prediction
cand = Path(work_dir) / "9_ML_Prediction"
if cand.is_dir():
paths.extend(sorted(p for p in cand.glob("*.csv") if p.is_file()))
return paths
def _resolve_geotiff_paths(config: Dict[str, Any], work_dir: str) -> List[Path]:
"""解析水色指数 GeoTIFF 路径列表（支持单文件 / 文件夹批量）"""
paths: List[Path] = []
single = config.get("geotiff_path")
folder = config.get("geotiff_dir")
if single:
p = Path(single)
if p.is_file():
paths.append(p)
if folder:
d = Path(folder)
if d.is_dir():
paths.extend(
sorted(p for p in d.glob("*.tif") if p.is_file())
)
paths.extend(
sorted(p for p in d.glob("*.TIF") if p.is_file())
)
if not paths and work_dir:
# 兜底：自动扫 10_WaterIndex_Images
cand = Path(work_dir) / "10_WaterIndex_Images"
if cand.is_dir():
paths.extend(sorted(p for p in cand.glob("*.tif") if p.is_file()))
paths.extend(sorted(p for p in cand.glob("*.TIF") if p.is_file()))
return paths
def _resolve_ref_image(work_dir: str) -> Optional[Path]:
"""自动找一张参考影像（用于克里金 CSV 模式的投影信息）"""
if not work_dir:
return None
for sub in ("3_deglint", "deglint", "10_WaterIndex_Images"):
d = Path(work_dir) / sub
if not d.is_dir():
continue
for pat in ("*.tif", "*.TIF", "*.bsq", "*.dat"):
cands = sorted(d.glob(pat))
if cands:
return cands[0]
return None
def _run_csv_mode(config: Dict[str, Any], csv_paths: List[Path],
output_dir: Path, work_dir: str) -> int:
"""CSV 插值模式：批量调用 batch_kriging_interpolation
Returns:
成功生成的文件数
"""
from src.utils.kriging import batch_kriging_interpolation
# 把 CSV 临时移到同一文件夹以便 batch_kriging_interpolation 处理
# （API 是 input_folder 而不是 list，这里直接调用一次覆盖全部）
if not csv_paths:
return 0
csv_dir = csv_paths[0].parent
# 找参考影像
ref_img = _resolve_ref_image(work_dir)
if ref_img is None:
# CSV 模式必须要有参考影像拿投影
raise FileNotFoundError(
"CSV 插值模式需要参考影像以读取投影信息，"
"未在工作目录找到 3_deglint/10_WaterIndex_Images 下的 GeoTIFF/BSQ"
)
resolution: float = float(config.get("resolution", 30.0))
print(f"[Step11 Service] CSV 插值模式: csv_dir={csv_dir}, ref={ref_img}")
print(f"[Step11 Service] resolution={resolution}m, output={output_dir}")
# batch_kriging_interpolation 在 output_dir 中生成 *{name}_kriging.tif
batch_kriging_interpolation(
input_folder=str(csv_dir),
ref_img_path=str(ref_img),
output_folder=str(output_dir),
spatial_resolution=resolution,
file_pattern="*.csv",
)
# 统计生成成功的 .tif 数
n_out = sum(1 for _ in output_dir.glob("*.tif"))
return n_out
def _run_geotiff_mode(geotiff_paths: List[Path],
output_dir: Path,
boundary_shp_path: Optional[str]) -> int:
"""GeoTIFF 栅格模式：ContentMapper.visualize_raster 逐个渲染为 PNG
Returns:
成功生成的 PNG 数
"""
import matplotlib
matplotlib.use("Agg") # 无头后端，service 内强制切换
from src.postprocessing.map import ContentMapper
mapper = ContentMapper()
output_dir.mkdir(parents=True, exist_ok=True)
n_ok = 0
for tif_path in geotiff_paths:
stem = tif_path.stem
chinese_title = mapper._get_chinese_title(stem)
output_png = output_dir / f"{chinese_title}_专题图.png"
try:
mapper.visualize_raster(
raster_tif_path=str(tif_path),
output_file=str(output_png),
boundary_shp_path=boundary_shp_path,
nodata_value=-9999.0,
figsize=(14, 10),
alpha=0.9,
)
n_ok += 1
print(f"[Step11 Service] ✅ {stem} → {output_png.name}")
except Exception as vis_err: # noqa: BLE001
print(f"[Step11 Service] ⚠ {stem} 渲染失败，跳过: {vis_err}")
continue
return n_ok
def execute_step11(config: Dict[str, Any]) -> Dict[str, Any]:
"""Step 11 后端计算入口——纯函数"""
# ---------- 入参规整 ----------
render_mode: str = config.get("render_mode", "CSV 插值模式")
boundary_shp_path: str = config.get("boundary_shp_path") or ""
enabled: bool = bool(config.get("enabled", True))
output_dir: str = config.get("output_dir") or ""
work_dir: str = config.get("work_dir") or "."
output_path = _resolve_output_dir(output_dir, work_dir)
mode = "make_thematic_map"
# ---------- 提前失败检查 ----------
if not enabled:
return {
"status": "skipped",
"output_path": None,
"message": "用户禁用此步骤（enabled=False）",
"mode": mode,
}
is_csv_mode = (render_mode == "CSV 插值模式")
if is_csv_mode:
csv_paths = _resolve_csv_paths(config, work_dir)
if not csv_paths:
return {
"status": "error",
"output_path": None,
"message": "未提供预测 CSV（prediction_csv_path / prediction_csv_dir）",
"mode": mode,
}
else:
geotiff_paths = _resolve_geotiff_paths(config, work_dir)
if not geotiff_paths:
return {
"status": "error",
"output_path": None,
"message": "未提供水色指数 GeoTIFF（geotiff_path / geotiff_dir）",
"mode": mode,
}
output_path.mkdir(parents=True, exist_ok=True)
# ---------- 执行 ----------
try:
print(f"[Step11 Service] render_mode={render_mode}")
if is_csv_mode:
n_out = _run_csv_mode(config, csv_paths, output_path, work_dir)
subdir_label = "克里金插值栅格"
else:
n_out = _run_geotiff_mode(
geotiff_paths,
output_path,
boundary_shp_path or None,
)
subdir_label = "GeoTIFF 渲染 PNG"
except FileNotFoundError as e:
return {
"status": "error",
"output_path": None,
"message": f"文件不存在: {e}",
"mode": mode,
}
except ValueError as e:
return {
"status": "error",
"output_path": None,
"message": f"参数错误: {e}",
"mode": mode,
}
except Exception as e: # noqa: BLE001
return {
"status": "error",
"output_path": None,
"message": f"{type(e).__name__}: {e}",
"mode": mode,
}
p = Path(output_path)
return {
"status": "completed",
"output_path": str(p).replace("\\", "/"),
"message": f"专题图完成（{subdir_label}），共 {n_out} 个输出文件，目录: {p.name or p}",
"mode": mode,
}

270
src/new/services/step12_service.py Normal file
View File

@ -0,0 +1,270 @@
# -*- coding: utf-8 -*-
"""
Step12 后端计算服务（数据可视化——散点/光谱/箱线/掩膜缩略/采样地图）
================================================================
纯计算函数——绝对不引用 PyQt、绝对不引用 main_view。它只：
1. 从 ``config`` 字典读取 ``generate_*`` 5 个开关；
2. 按开关依次调用 ``src/core/visualization`` 或
``src/postprocessing.visualization_reports`` 中的独立生成函数；
3. 每个子任务独立 try/except 隔离，单个失败不影响其它；
4. 返回结果字典 ``{status, output_path, message, mode}``。
调用入口（由 main_view 在后台 QThread 中调用）：
execute_step12({
"work_dir": "D:/workspace", # 工作目录（必填）
"img_dir": "D:/workspace/9_ML_Prediction", # 图像目录（可省，自动推断）
"generate_scatter": True, # 模型评估散点图
"generate_spectrum": True, # 光谱曲线图
"generate_boxplots": True, # 箱线图
"generate_glint_previews": True, # 掩膜/耀斑缩略图
"generate_sampling_maps": True, # 采样点地图
"models_dir": "D:/8_Supervised_Model_Training", # 散点图依赖（可省，自动推断）
"training_csv_path": "D:/7_Water_Quality_Indices/training_with_indices.csv", # 光谱/箱线依赖（可省，自动推断）
"output_dir": "D:/14_visualization", # 输出目录（可省 → work_dir/14_visualization）
"enabled": True,
})
返回字典字段：
* ``status`` : "completed" | "skipped" | "error"
* ``output_path`` : 输出目录路径（失败时为 None）
* ``message`` : 人类可读说明
* ``mode`` : "viz_generate"（便于 UI 提示）
设计取舍
--------
- 旧 panel 把所有图嵌入 matplotlib 画布；service 端只生成 PNG 文件，
完全离线——view 层不再嵌入任何 chart widget。
- 单个可视化失败时 ``results[sub] = {"status": "error", ...}``，
但外层 status 仍然 = "completed"（局部失败不等于全部失败）。
- 完全没有可用数据（连 work_dir 都不存在）时才返回 status="error"。
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict
def _resolve_output_dir(output_dir: str | None, work_dir: str) -> Path:
if output_dir:
return Path(output_dir)
return Path(work_dir) / "14_visualization"
def _resolve_models_dir(work_dir: str, models_dir: str | None) -> str:
"""自动推断模型目录"""
if models_dir and Path(models_dir).is_dir():
return models_dir
cand = Path(work_dir) / "8_Supervised_Model_Training"
return str(cand) if cand.is_dir() else (models_dir or "")
def _resolve_training_csv(work_dir: str, training_csv_path: str | None) -> str:
"""自动推断训练 CSV"""
if training_csv_path and Path(training_csv_path).is_file():
return training_csv_path
for sub in (
"7_Water_Quality_Indices",
"6_Spectral_Feature_Extraction",
"4_processed_data",
"visualization",
):
d = Path(work_dir) / sub
if not d.is_dir():
continue
cands = sorted(d.glob("*.csv"))
if cands:
return str(cands[0])
return training_csv_path or ""
def _try_scatter(work_dir: str, output_dir: Path) -> Dict[str, Any]:
"""生成模型评估散点图（依赖 models_dir + training_csv_path）"""
from src.core.visualization.scatter_plot import generate_model_scatter_plots
models_dir = _resolve_models_dir(work_dir, None)
training_csv = _resolve_training_csv(work_dir, None)
if not models_dir or not Path(models_dir).is_dir():
raise FileNotFoundError(f"模型目录不存在: {models_dir or '(自动推断失败)'}")
if not training_csv or not Path(training_csv).is_file():
raise FileNotFoundError(f"训练 CSV 不存在: {training_csv or '(自动推断失败)'}")
out_sub = output_dir / "scatter_plots"
paths = generate_model_scatter_plots(
models_dir=models_dir,
training_csv_path=training_csv,
output_dir=str(out_sub),
)
return {"status": "completed", "count": len(paths), "output_dir": str(out_sub)}
def _try_spectrum(work_dir: str, output_dir: Path) -> Dict[str, Any]:
"""生成光谱曲线对比图（依赖 training CSV）"""
from src.core.visualization.spectrum_plot import generate_spectrum_comparison_plots
training_csv = _resolve_training_csv(work_dir, None)
if not training_csv or not Path(training_csv).is_file():
raise FileNotFoundError(f"训练 CSV 不存在: {training_csv or '(自动推断失败)'}")
out_sub = output_dir / "spectrum_plots"
paths = generate_spectrum_comparison_plots(
csv_path=training_csv,
output_dir=str(out_sub),
)
return {"status": "completed", "count": len(paths), "output_dir": str(out_sub)}
def _try_boxplots(work_dir: str, output_dir: Path) -> Dict[str, Any]:
"""生成水质参数箱型图（依赖 training CSV）"""
from src.core.visualization.boxplot import generate_boxplots
training_csv = _resolve_training_csv(work_dir, None)
if not training_csv or not Path(training_csv).is_file():
raise FileNotFoundError(f"训练 CSV 不存在: {training_csv or '(自动推断失败)'}")
out_sub = output_dir / "boxplots"
paths = generate_boxplots(
csv_path=training_csv,
output_dir=str(out_sub),
)
return {"status": "completed", "count": len(paths), "output_dir": str(out_sub)}
def _try_glint_previews(work_dir: str, output_dir: Path) -> Dict[str, Any]:
"""生成耀斑分析影像预览图"""
from src.postprocessing.visualization_reports import ReportGenerator
rg = ReportGenerator(output_dir=str(output_dir))
paths = rg.generate_glint_deglint_previews(
work_dir=work_dir,
output_subdir="glint_deglint_previews",
generate_glint=True,
generate_deglint=True,
)
return {"status": "completed", "count": len(paths), "output_dir": str(output_dir / "glint_deglint_previews")}
def _try_sampling_maps(work_dir: str, output_dir: Path) -> Dict[str, Any]:
"""生成采样点地图"""
from src.postprocessing.visualization_reports import ReportGenerator
rg = ReportGenerator(output_dir=str(output_dir))
p = rg.generate_sampling_point_map(output_subdir="sampling_maps")
return {"status": "completed" if p else "error", "path": p, "output_dir": str(output_dir / "sampling_maps")}
def execute_step12(config: Dict[str, Any]) -> Dict[str, Any]:
"""Step 12 后端计算入口——纯函数"""
work_dir: str = config.get("work_dir") or ""
img_dir: str = config.get("img_dir") or ""
enabled: bool = bool(config.get("enabled", True))
output_dir: str = config.get("output_dir") or ""
# 5 个开关：缺省默认 True（与旧 panel 行为一致）
gen_scatter = bool(config.get("generate_scatter", True))
gen_spectrum = bool(config.get("generate_spectrum", True))
gen_boxplots = bool(config.get("generate_boxplots", True))
gen_glint = bool(config.get("generate_glint_previews", True))
gen_sampling = bool(config.get("generate_sampling_maps", True))
output_path = _resolve_output_dir(output_dir, work_dir)
mode = "viz_generate"
# ---------- 提前失败检查 ----------
if not enabled:
return {
"status": "skipped",
"output_path": None,
"message": "用户禁用此步骤（enabled=False）",
"mode": mode,
}
if not work_dir:
return {
"status": "error",
"output_path": None,
"message": "未提供工作目录（work_dir）",
"mode": mode,
}
if not Path(work_dir).is_dir():
return {
"status": "error",
"output_path": None,
"message": f"工作目录不存在: {work_dir}",
"mode": mode,
}
output_path.mkdir(parents=True, exist_ok=True)
# img_dir 仅作提示用，不强制校验
_ = img_dir
# ---------- 逐项执行（独立 try/except 隔离） ----------
results: Dict[str, Dict[str, Any]] = {}
n_ok = 0
n_skip = 0
n_err = 0
tasks = []
if gen_scatter:
tasks.append(("scatter", _try_scatter))
if gen_spectrum:
tasks.append(("spectrum", _try_spectrum))
if gen_boxplots:
tasks.append(("boxplots", _try_boxplots))
if gen_glint:
tasks.append(("glint_previews", _try_glint_previews))
if gen_sampling:
tasks.append(("sampling_maps", _try_sampling_maps))
if not tasks:
return {
"status": "completed",
"output_path": str(output_path).replace("\\", "/"),
"message": "无可视化任务（5 个开关全部 False）",
"mode": mode,
}
print(f"[Step12 Service] 工作目录: {work_dir}")
print(f"[Step12 Service] 输出目录: {output_path}")
print(f"[Step12 Service] 子任务: {[n for n, _ in tasks]}")
for name, fn in tasks:
try:
r = fn(work_dir, output_path)
results[name] = r
if r.get("status") == "completed":
n_ok += 1
print(f" ✅ {name}: count={r.get('count', r.get('path', '?'))}")
else:
n_err += 1
print(f" ⚠ {name}: status={r.get('status')}")
except FileNotFoundError as e:
results[name] = {"status": "skipped", "reason": str(e)}
n_skip += 1
print(f" ↻ {name}: 跳过（缺数据）: {e}")
except Exception as e: # noqa: BLE001
results[name] = {"status": "error", "message": f"{type(e).__name__}: {e}"}
n_err += 1
print(f" ❌ {name}: {type(e).__name__}: {e}")
# ---------- 汇总 ----------
# 局部失败不算全局失败——只有当所有任务都缺数据时才算 skipped
if n_ok == 0 and n_err == 0:
return {
"status": "skipped",
"output_path": None,
"message": f"全部 {n_skip} 个子任务缺数据被跳过（models_dir/training_csv 等前置产物不存在）",
"mode": mode,
}
return {
"status": "completed",
"output_path": str(output_path).replace("\\", "/"),
"message": (
f"可视化完成：成功 {n_ok} / 失败 {n_err} / 跳过 {n_skip}；"
f"目录: {output_path.name or output_path}"
),
"mode": mode,
}

221
src/new/services/step13_service.py Normal file
View File

@ -0,0 +1,221 @@
# -*- coding: utf-8 -*-
"""
Step13 后端计算服务（Word 报告生成）
====================================
纯计算函数——绝对不引用 PyQt、绝对不引用 main_view。它只：
1. 从 ``config`` 字典读取参数；
2. 调用 ``WaterQualityReportGenerator.generate_report`` 把工作目录
下的可视化结果（14_visualization 等）拼装成 Word 文档；
3. AI 配置（Provider / API Key / Model / Timeout）从环境变量读取，
与 ``ReportGenerationConfig`` 默认行为完全一致；调用方可在 dispatch
前将 QSettings 内容写入环境变量（如 ``AI_PROVIDER`` / ``MINIMAX_API_KEY``）；
4. 返回结果字典 ``{status, output_path, message, mode}``。
调用入口（由 main_view 在后台 QThread 中调用）：
execute_step13({
"work_dir": "D:/workspace", # 工作目录（必填）
"output_dir": "D:/workspace/14_visualization", # 输出目录（可省 → work_dir/14_visualization）
"report_title": "水质参数反演分析报告", # 报告标题
"enable_ai_analysis": True, # 是否启用 AI 解读
# --- AI 配置（可省；缺省从环境变量 AI_PROVIDER / MINIMAX_API_KEY 等读） ---
"ai_provider": "minimax" | "ollama",
"ai_api_key": "...",
"ai_api_base_url": "...",
"ai_vision_model": "...",
"ai_text_model": "...",
"ai_timeout_s": 120,
"enabled": True,
})
返回字典字段：
* ``status`` : "completed" | "skipped" | "error"
* ``output_path`` : 生成的 .docx 文件路径
* ``message`` : 人类可读说明
* ``mode`` : "generate_word_report"（便于 UI 提示）
设计取舍
--------
- AI 配置优先用 config 字典传入（与新架构保持一致，避免在 service 内
引用 Qt 的 QSettings）；缺省时 ReportGenerator 内部从环境变量兜底。
- view 层如有 QSettings 配置，dispatch 前需 main_view 把 settings 写到 env
（或直接塞进 config dict）。
- 旧 panel 把 QSettings 在 worker 内部读，跨线程不安全；新架构让主线程
准备好数据后再丢进 TaskWorker。
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Dict, Optional
def _resolve_output_dir(output_dir: Optional[str], work_dir: str) -> Path:
"""根据 output_dir / work_dir 计算 Word 报告输出目录"""
if output_dir:
return Path(output_dir)
return Path(work_dir) / "14_visualization"
def _apply_ai_env(config: Dict[str, Any]) -> None:
"""把 config 里的 AI 配置写到环境变量（如果给了），让 ReportGenerator 内部兜底
注意：直接修改 os.environ 是进程级副作用，但 ReportGenerationConfig 已用
环境变量兜底——这是复用既有逻辑的最简方式。
"""
mapping = {
"ai_provider": "AI_PROVIDER",
"ai_api_key": "MINIMAX_API_KEY",
"ai_api_base_url": "MINIMAX_BASE_URL", # 仅 Minimax 用
"ai_vision_model": None, # 见下：根据 provider 分流
"ai_text_model": None,
"ai_timeout_s": "MINIMAX_TIMEOUT_S",
}
provider = config.get("ai_provider") or os.environ.get("AI_PROVIDER", "")
provider = (provider or "").lower() or "minimax"
for cfg_key, env_key in mapping.items():
val = config.get(cfg_key)
if val is None or val == "":
continue
if env_key is None:
# vision / text model：根据 provider 分流
if cfg_key == "ai_vision_model":
env_key = (
"OLLAMA_VISION_MODEL" if provider == "ollama"
else "MINIMAX_VISION_MODEL"
)
elif cfg_key == "ai_text_model":
env_key = (
"OLLAMA_TEXT_MODEL" if provider == "ollama"
else "MINIMAX_TEXT_MODEL"
)
if env_key:
os.environ[env_key] = str(val)
if provider == "ollama":
os.environ.setdefault("AI_PROVIDER", "ollama")
elif provider == "minimax":
os.environ.setdefault("AI_PROVIDER", "minimax")
# ENABLE_AI_ANALYSIS 由 ReportGenerator 用 0/false 字符串判断
if "enable_ai_analysis" in config:
os.environ["ENABLE_AI_ANALYSIS"] = "1" if bool(config["enable_ai_analysis"]) else "0"
def execute_step13(config: Dict[str, Any]) -> Dict[str, Any]:
"""Step 13 后端计算入口——纯函数"""
# ---------- 入参规整 ----------
work_dir: str = config.get("work_dir") or ""
output_dir: str = config.get("output_dir") or ""
report_title: str = (
config.get("report_title") or "水质参数反演分析报告"
).strip()
enabled: bool = bool(config.get("enabled", True))
output_path = _resolve_output_dir(output_dir, work_dir)
mode = "generate_word_report"
# ---------- 提前失败检查 ----------
if not enabled:
return {
"status": "skipped",
"output_path": None,
"message": "用户禁用此步骤（enabled=False）",
"mode": mode,
}
if not work_dir:
return {
"status": "error",
"output_path": None,
"message": "未提供工作目录（work_dir）",
"mode": mode,
}
if not Path(work_dir).is_dir():
return {
"status": "error",
"output_path": None,
"message": f"工作目录不存在: {work_dir}",
"mode": mode,
}
vis_dir = Path(work_dir) / "14_visualization"
if not vis_dir.is_dir():
return {
"status": "error",
"output_path": None,
"message": (
f"可视化目录不存在: {vis_dir}。"
"请先执行 Step 12 生成可视化图表。"
),
"mode": mode,
}
# ---------- 注入 AI 环境变量 ----------
_apply_ai_env(config)
# ---------- 执行 ----------
try:
from src.postprocessing.report_word import (
WaterQualityReportGenerator,
ReportGenerationConfig,
)
ai_cfg = ReportGenerationConfig(
ai_provider=config.get("ai_provider") or os.environ.get("AI_PROVIDER", "minimax"),
enable_ai_analysis=bool(config.get("enable_ai_analysis", True)),
minimax_api_key=config.get("ai_api_key") or os.environ.get("MINIMAX_API_KEY", ""),
minimax_vision_model=config.get("ai_vision_model") or None,
minimax_text_model=config.get("ai_text_model") or None,
ollama_base_url=config.get("ai_api_base_url") or None,
ollama_vision_model=config.get("ai_vision_model") or None,
ollama_text_model=config.get("ai_text_model") or None,
)
print(f"[Step13 Service] 工作目录: {work_dir}")
print(f"[Step13 Service] 输出目录: {output_path}")
print(f"[Step13 Service] 报告标题: {report_title}")
print(f"[Step13 Service] AI 分析: {'开' if ai_cfg.enable_ai_analysis else '关'}")
gen = WaterQualityReportGenerator(
output_dir=str(output_path),
work_dir=work_dir,
ai_config=ai_cfg,
)
out_path = gen.generate_report(
work_dir=work_dir,
report_title=report_title,
)
except FileNotFoundError as e:
return {
"status": "error",
"output_path": None,
"message": f"文件不存在: {e}",
"mode": mode,
}
except ValueError as e:
return {
"status": "error",
"output_path": None,
"message": f"参数错误: {e}",
"mode": mode,
}
except Exception as e: # noqa: BLE001
return {
"status": "error",
"output_path": None,
"message": f"{type(e).__name__}: {e}",
"mode": mode,
}
# ---------- 成功路径 ----------
return {
"status": "completed",
"output_path": str(out_path).replace("\\", "/"),
"message": f"Word 报告已生成: {Path(out_path).name}",
"mode": mode,
}