110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
成图步骤
|
||
|
||
包含 step9_generate_distribution_map
|
||
"""
|
||
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Optional, Union, Callable
|
||
|
||
|
||
class MappingStep:
|
||
"""成图步骤"""
|
||
|
||
@staticmethod
|
||
def generate_distribution_map(
|
||
prediction_csv_path: str,
|
||
boundary_shp_path: str,
|
||
output_image_path: Optional[str] = None,
|
||
resolution: float = 30,
|
||
input_crs: str = "EPSG:32651",
|
||
output_crs: str = "EPSG:4326",
|
||
show_sample_points: bool = False,
|
||
base_map_tif: Optional[str] = None,
|
||
use_distance_diffusion: bool = True,
|
||
max_diffusion_distance: Optional[float] = None,
|
||
diffusion_power: float = 2,
|
||
diffusion_n_neighbors: int = 15,
|
||
cmap: Optional[str] = None,
|
||
expand_ratio: float = 0.05,
|
||
output_dir: Union[str, Path] = "./14_visualization",
|
||
callback: Optional[Callable] = None,
|
||
) -> str:
|
||
"""
|
||
根据采样点的坐标和反演的实测参数，通过插值方法得到水质参数可视化分布图
|
||
|
||
Args:
|
||
prediction_csv_path: 预测结果CSV文件路径（前两列为经纬度，第三列为预测值）
|
||
boundary_shp_path: 边界shapefile文件路径
|
||
output_image_path: 输出图片路径（如果为None，自动生成）
|
||
resolution: 插值网格分辨率（米）
|
||
input_crs: 输入坐标系
|
||
output_crs: 输出坐标系
|
||
show_sample_points: 是否在图上显示采样点
|
||
base_map_tif: 底图TIF路径
|
||
use_distance_diffusion: 是否启用距离扩散补全边界
|
||
max_diffusion_distance: 距离扩散最大距离（米）
|
||
diffusion_power: 距离扩散幂参数
|
||
diffusion_n_neighbors: 距离扩散最近邻数量
|
||
cmap: 颜色映射名称（None表示自动识别）
|
||
expand_ratio: 边界外扩比例（0-1之间）
|
||
output_dir: 输出目录
|
||
callback: 回调函数
|
||
|
||
Returns:
|
||
可视化分布图文件路径
|
||
"""
|
||
from src.postprocessing.map import ContentMapper
|
||
|
||
output_dir = Path(output_dir)
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
def notify(status, msg=""):
|
||
if callback:
|
||
callback("步骤9", status, msg)
|
||
|
||
print("\n" + "=" * 80)
|
||
print("步骤9: 生成水质参数可视化分布图")
|
||
print("=" * 80)
|
||
|
||
step_start_time = time.time()
|
||
|
||
if output_image_path is None:
|
||
csv_name = Path(prediction_csv_path).stem
|
||
output_image_path = str(output_dir / f"{csv_name}_distribution.png")
|
||
|
||
if Path(output_image_path).exists():
|
||
print(f"检测到已存在的分布图文件，直接使用: {output_image_path}")
|
||
notify("skipped", f"可视化分布图已设置: {output_image_path}")
|
||
return output_image_path
|
||
|
||
mapper = ContentMapper(input_crs=input_crs, output_crs=output_crs)
|
||
|
||
mapper_kwargs = {
|
||
"resolution": resolution,
|
||
"show_sample_points": show_sample_points,
|
||
"use_distance_diffusion": use_distance_diffusion,
|
||
"diffusion_power": diffusion_power,
|
||
"diffusion_n_neighbors": diffusion_n_neighbors,
|
||
"expand_ratio": expand_ratio,
|
||
}
|
||
|
||
optional_kwargs = {
|
||
"base_map_tif": base_map_tif,
|
||
"max_diffusion_distance": max_diffusion_distance,
|
||
"cmap": cmap,
|
||
}
|
||
mapper_kwargs.update({k: v for k, v in optional_kwargs.items() if v is not None})
|
||
|
||
mapper.process_data(
|
||
csv_file=prediction_csv_path,
|
||
shp_file=boundary_shp_path,
|
||
output_file=output_image_path,
|
||
**mapper_kwargs,
|
||
)
|
||
|
||
notify("completed", f"可视化分布图已保存: {output_image_path}")
|
||
return output_image_path
|