fix: 补齐缺失的 handler 文件 + IDW 插值退化检测 + V1 代码归档
This commit is contained in:
duxin
99
src/core/handlers/step10_watercolor_handler.py
Normal file
99
src/core/handlers/step10_watercolor_handler.py
Normal file
@ -0,0 +1,99 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Step10 处理器：水色指数反演（CSV 散点模式）
|
||||
|
||||
2026-06-30 新建：替代原先错误路由的 step10_qaa（QAA 物理反演）。
|
||||
使用 WaterIndexCsvProcessor 对采样点 CSV 逐行计算选定的水质指数公式，
|
||||
每个公式输出一个独立 CSV 文件（longitude, latitude, <公式值>）。
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from src.core.handlers.base import BaseStepHandler, PipelineContext
|
||||
|
||||
|
||||
class Step10WatercolorHandler(BaseStepHandler):
|
||||
"""步骤10：水色指数反演（CSV 散点模式）。
|
||||
|
||||
对应 config key: 'step10_watercolor'
|
||||
委托类: WaterIndexCsvProcessor
|
||||
"""
|
||||
|
||||
step_key = 'step10_watercolor'
|
||||
|
||||
def execute(self, context: PipelineContext, config: dict) -> Dict[str, Any]:
|
||||
from src.core.algorithms.waterindex_inversion.csv_processor import (
|
||||
WaterIndexCsvProcessor,
|
||||
)
|
||||
|
||||
step_start_time = time.time()
|
||||
|
||||
# ── 解析参数 ──
|
||||
sampling_csv_path = config.get('sampling_csv_path', '')
|
||||
if not sampling_csv_path or not os.path.isfile(sampling_csv_path):
|
||||
msg = f'采样点 CSV 不存在或路径为空: {sampling_csv_path}'
|
||||
context.notify('step10_watercolor', 'error', msg)
|
||||
return {'error': msg}
|
||||
|
||||
output_dir = config.get('output_dir', '')
|
||||
if not output_dir:
|
||||
output_dir = os.path.join(str(context.work_dir), '10_WaterIndex_CSV')
|
||||
|
||||
# 显式兜底：只有非空 list 才启用过滤，杜绝 falsy 空列表透传
|
||||
raw_selected = config.get('selected_formulas')
|
||||
selected_formulas: Optional[list] = (
|
||||
raw_selected if isinstance(raw_selected, list) and len(raw_selected) > 0 else None
|
||||
)
|
||||
wavelength_offset = float(config.get('wavelength_offset', 0))
|
||||
waterindex_csv = config.get('waterindex_csv', '')
|
||||
|
||||
if not waterindex_csv or not os.path.isfile(waterindex_csv):
|
||||
msg = f'waterindex.csv 不存在（前端未传入或路径无效）: {waterindex_csv!r}'
|
||||
context.notify('step10_watercolor', 'error', msg)
|
||||
return {'error': msg}
|
||||
|
||||
# ── 执行 ──
|
||||
try:
|
||||
processor = WaterIndexCsvProcessor(waterindex_csv)
|
||||
|
||||
from src.gui.core.event_bus import global_event_bus
|
||||
|
||||
def _progress_callback(msg: str, pct: float):
|
||||
context.notify('step10_watercolor', 'info', f'[{pct:.0f}%] {msg}')
|
||||
global_event_bus.publish('ProgressUpdate', {
|
||||
'percentage': int(pct),
|
||||
'message': f'Step10 水色指数: {msg}',
|
||||
})
|
||||
|
||||
out_files = processor.compute_indices_from_csv(
|
||||
sampling_csv_path=sampling_csv_path,
|
||||
output_dir=output_dir,
|
||||
selected_formulas=selected_formulas,
|
||||
progress_callback=_progress_callback,
|
||||
wavelength_offset=wavelength_offset,
|
||||
)
|
||||
|
||||
step_end_time = time.time()
|
||||
context.record_step_time(
|
||||
'步骤10: 水色指数反演', step_start_time, step_end_time
|
||||
)
|
||||
context.notify(
|
||||
'step10_watercolor', 'completed',
|
||||
f'水色指数反演完成，共生成 {len(out_files)} 个 CSV'
|
||||
)
|
||||
|
||||
return {
|
||||
'output_dir': output_dir,
|
||||
'output_files': out_files,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
step_end_time = time.time()
|
||||
context.record_step_time(
|
||||
'步骤10: 水色指数反演', step_start_time, step_end_time,
|
||||
status='failed', error=str(e)
|
||||
)
|
||||
raise
|
||||
268
src/core/handlers/step11_map_handler.py
Normal file
268
src/core/handlers/step11_map_handler.py
Normal file
@ -0,0 +1,268 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Step11 处理器：专题图生成（CSV 插值 → 分布图 PNG）
|
||||
|
||||
2026-06-30：统一到 Pipeline 体系。
|
||||
- 预矢量化：将 .dat 栅格掩膜转一次为 .shp，避免 63 个 CSV 各自矢量化 1765 万像元。
|
||||
- 串行处理：GDAL 不具有线程安全性，多线程并发 Open/Polygonize 导致 0xC0000005。
|
||||
预矢量化后每个 CSV 处理时间大幅缩短，串行速度已可接受。
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
from src.core.handlers.base import BaseStepHandler, PipelineContext
|
||||
|
||||
|
||||
def _process_one_map(csv_path: str, base_kwargs: dict, output_dir: str
|
||||
) -> Tuple[str, str]:
|
||||
"""处理单个 CSV → 分布图 TIF（子进程内执行）。
|
||||
|
||||
2026-07-01：支持 shared_context 快速通道。
|
||||
当 base_kwargs 包含 'shared_context' 时，绕过 MappingStep，
|
||||
直接用 ContentMapper.process_data(shared_context=ctx) 跳过重复的边界/网格/掩膜计算。
|
||||
"""
|
||||
import matplotlib
|
||||
try:
|
||||
matplotlib.use('Agg', force=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
stem = Path(csv_path).stem
|
||||
output_image_path = str(Path(output_dir) / f'{stem}_distribution.png')
|
||||
|
||||
shared_context = base_kwargs.get('shared_context')
|
||||
|
||||
if shared_context is not None:
|
||||
# ★ 快速通道：直接调 ContentMapper，复用预计算的网格/掩膜
|
||||
from src.postprocessing.map import ContentMapper
|
||||
mapper = ContentMapper(
|
||||
input_crs=base_kwargs.get('input_crs', 'EPSG:32651'),
|
||||
output_crs=base_kwargs.get('output_crs', base_kwargs.get('input_crs', 'EPSG:32651')),
|
||||
)
|
||||
result_path = mapper.process_data(
|
||||
csv_file=csv_path,
|
||||
shp_file=base_kwargs.get('boundary_shp_path'),
|
||||
output_file=output_image_path,
|
||||
resolution=float(base_kwargs.get('resolution', 10.0)),
|
||||
output_format='tif',
|
||||
shared_context=shared_context,
|
||||
)
|
||||
return result_path, csv_path
|
||||
|
||||
# 原有路径（无共享上下文时）
|
||||
kw = dict(base_kwargs)
|
||||
kw.pop('shared_context', None) # 防止透传给 MappingStep
|
||||
kw['prediction_csv_path'] = csv_path
|
||||
kw['output_image_path'] = output_image_path
|
||||
|
||||
from src.core.steps.mapping_step import MappingStep
|
||||
result_path = MappingStep.generate_distribution_map(**kw)
|
||||
return result_path, csv_path
|
||||
|
||||
|
||||
def _pre_vectorize_boundary(boundary_path: str, work_dir: str) -> str:
|
||||
"""将栅格掩膜预转换为矢量 SHP（只做一次），返回 SHP 路径。"""
|
||||
src = Path(boundary_path)
|
||||
if not src.exists():
|
||||
return boundary_path
|
||||
if src.suffix.lower() in ('.shp', '.geojson'):
|
||||
return boundary_path
|
||||
|
||||
cache_dir = Path(work_dir) / '.step11_cache'
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
shp_out = cache_dir / f'{src.stem}_vectorized.shp'
|
||||
if shp_out.exists():
|
||||
return str(shp_out)
|
||||
|
||||
from osgeo import gdal, ogr, osr
|
||||
gdal.UseExceptions()
|
||||
|
||||
ds = gdal.Open(str(src))
|
||||
if ds is None:
|
||||
return boundary_path
|
||||
|
||||
band = ds.GetRasterBand(1)
|
||||
drv = ogr.GetDriverByName('ESRI Shapefile')
|
||||
if drv is None:
|
||||
ds = None
|
||||
return boundary_path
|
||||
|
||||
for ext in ('.shp', '.shx', '.dbf', '.prj'):
|
||||
p = cache_dir / f'{src.stem}_vectorized{ext}'
|
||||
if p.exists():
|
||||
p.unlink()
|
||||
|
||||
out_ds = drv.CreateDataSource(str(shp_out))
|
||||
srs = osr.SpatialReference()
|
||||
proj = ds.GetProjection()
|
||||
if proj:
|
||||
srs.ImportFromWkt(proj)
|
||||
else:
|
||||
srs.ImportFromEPSG(32651)
|
||||
|
||||
layer = out_ds.CreateLayer('boundary', srs, ogr.wkbMultiPolygon)
|
||||
gdal.Polygonize(band, band, layer, 0, [], callback=None)
|
||||
|
||||
out_ds = None
|
||||
ds = None
|
||||
|
||||
if shp_out.exists():
|
||||
return str(shp_out)
|
||||
return boundary_path
|
||||
|
||||
|
||||
class Step11MapHandler(BaseStepHandler):
|
||||
"""步骤11：专题图生成（预矢量化 + 串行）。
|
||||
|
||||
对应 config key: 'step11_map'
|
||||
"""
|
||||
|
||||
step_key = 'step11_map'
|
||||
|
||||
def execute(self, context: PipelineContext, config: dict) -> Dict[str, Any]:
|
||||
step_start_time = time.time()
|
||||
|
||||
prediction_csv_path = config.get('prediction_csv_path', '')
|
||||
prediction_csv_dir = config.get('prediction_csv_dir', '')
|
||||
batch_mode = config.get('batch_mode', False)
|
||||
boundary_shp_path = config.get('boundary_shp_path', '') or ''
|
||||
output_dir = config.get('output_dir', '') or str(context.visualization_dir)
|
||||
output_image_path = config.get('output_image_path', '')
|
||||
|
||||
# ── 收集 CSV ──
|
||||
csv_paths: List[str] = []
|
||||
if prediction_csv_path and os.path.isfile(prediction_csv_path):
|
||||
csv_paths = [prediction_csv_path]
|
||||
elif batch_mode and prediction_csv_dir and os.path.isdir(prediction_csv_dir):
|
||||
import glob
|
||||
csv_paths = sorted(glob.glob(os.path.join(prediction_csv_dir, '*.csv')))
|
||||
|
||||
if not csv_paths:
|
||||
context.notify('step11_map', 'error', '没有找到预测 CSV 文件')
|
||||
return {'error': '没有找到预测 CSV 文件'}
|
||||
|
||||
# ── ★ 预矢量化边界（仅一次）──
|
||||
resolved_boundary = boundary_shp_path
|
||||
if boundary_shp_path and os.path.isfile(boundary_shp_path):
|
||||
try:
|
||||
context.notify('step11_map', 'info', '正在预矢量化水域边界（仅一次）…')
|
||||
resolved_boundary = _pre_vectorize_boundary(
|
||||
boundary_shp_path, str(context.work_dir)
|
||||
)
|
||||
context.notify('step11_map', 'info',
|
||||
f'边界预处理完成，开始生成专题图…')
|
||||
except Exception as e:
|
||||
context.notify('step11_map', 'warning',
|
||||
f'边界预处理失败，使用原始路径: {e}')
|
||||
|
||||
base_kwargs = {
|
||||
'boundary_shp_path': resolved_boundary or None,
|
||||
'resolution': float(config.get('resolution', 10.0)),
|
||||
'input_crs': config.get('input_crs', 'EPSG:32651'),
|
||||
'output_crs': config.get('output_crs', config.get('input_crs', 'EPSG:32651')),
|
||||
'show_sample_points': bool(config.get('show_sample_points', False)),
|
||||
'use_distance_diffusion': bool(config.get('use_distance_diffusion', False)),
|
||||
}
|
||||
if output_image_path and len(csv_paths) == 1:
|
||||
base_kwargs['output_image_path'] = output_image_path
|
||||
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
|
||||
total = len(csv_paths)
|
||||
from src.gui.core.event_bus import global_event_bus
|
||||
|
||||
context.notify('step11_map', 'info', f'专题图开始（{total} 个 CSV）')
|
||||
global_event_bus.publish('ProgressUpdate', {
|
||||
'percentage': 0, 'message': f'Step11 专题图: 0/{total}',
|
||||
})
|
||||
|
||||
# ── ★ 2026-07-01：主进程预计算共享空间上下文 ──
|
||||
# 63 个 CSV 坐标一致，边界/网格/掩膜只算一次，塞入 base_kwargs
|
||||
# 子进程通过 shared_context 复用，跳过重复的 ②④⑥ (~1.2s/CSV)
|
||||
if total > 1 and resolved_boundary:
|
||||
try:
|
||||
context.notify('step11_map', 'info',
|
||||
'正在预计算共享空间上下文（边界+网格+掩膜，仅一次）…')
|
||||
from src.postprocessing.map import ContentMapper
|
||||
pre_mapper = ContentMapper(
|
||||
input_crs=base_kwargs['input_crs'],
|
||||
output_crs=base_kwargs['output_crs'],
|
||||
)
|
||||
shared_ctx = pre_mapper.prepare_shared_context(
|
||||
sample_csv=csv_paths[0],
|
||||
shp_file=resolved_boundary,
|
||||
resolution=float(base_kwargs['resolution']),
|
||||
expand_ratio=0.05,
|
||||
)
|
||||
base_kwargs['shared_context'] = shared_ctx
|
||||
context.notify('step11_map', 'info',
|
||||
f'✓ 共享上下文已就绪，后续 {total} 个 CSV 将直接复用')
|
||||
except Exception as e:
|
||||
context.notify('step11_map', 'warning',
|
||||
f'共享上下文预计算失败: {e}，回退逐个处理')
|
||||
|
||||
# ── 多进程并行（GDAL 线程不安全，但进程隔离下安全）──
|
||||
generated: List[str] = []
|
||||
errors: Dict[str, str] = {}
|
||||
|
||||
import multiprocessing
|
||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||||
|
||||
# 留出 1-2 个核心保证电脑不卡死
|
||||
max_workers = max(1, multiprocessing.cpu_count() - 2)
|
||||
context.notify('step11_map', 'info', f'使用 {max_workers} 个进程并行生成')
|
||||
|
||||
with ProcessPoolExecutor(max_workers=max_workers) as executor:
|
||||
future_to_csv = {
|
||||
executor.submit(_process_one_map, csv_p, base_kwargs, output_dir): csv_p
|
||||
for csv_p in csv_paths
|
||||
}
|
||||
done_count = 0
|
||||
for future in as_completed(future_to_csv):
|
||||
csv_p = future_to_csv[future]
|
||||
done_count += 1
|
||||
try:
|
||||
result_path, _ = future.result()
|
||||
generated.append(result_path)
|
||||
except Exception as e:
|
||||
errors[csv_p] = str(e)
|
||||
context.notify('step11_map', 'warning',
|
||||
f'专题图 FAIL: {Path(csv_p).name} — {e}')
|
||||
|
||||
pct = int(done_count / total * 100)
|
||||
global_event_bus.publish('ProgressUpdate', {
|
||||
'percentage': pct,
|
||||
'message': f'Step11 专题图: {done_count}/{total}',
|
||||
})
|
||||
if done_count % max(1, total // 10) == 0 or done_count == total:
|
||||
context.notify('step11_map', 'info',
|
||||
f'专题图 [{done_count}/{total}]')
|
||||
|
||||
step_end_time = time.time()
|
||||
elapsed = step_end_time - step_start_time
|
||||
context.record_step_time('步骤11: 专题图生成', step_start_time, step_end_time)
|
||||
|
||||
result_msg = f'专题图完成：{len(generated)}/{total} 张（{elapsed:.1f}s）'
|
||||
if errors:
|
||||
result_msg += f'，{len(errors)} 个失败'
|
||||
context.notify('step11_map', 'completed', result_msg)
|
||||
|
||||
# 清理临时缓存
|
||||
try:
|
||||
cache_dir = Path(str(context.work_dir)) / '.step11_cache'
|
||||
if cache_dir.exists():
|
||||
shutil.rmtree(cache_dir, ignore_errors=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return {
|
||||
'generated': generated,
|
||||
'errors': errors,
|
||||
'output_dir': output_dir,
|
||||
'elapsed_seconds': elapsed,
|
||||
}
|
||||
Reference in New Issue
Block a user