pyufim 1.0.1

API-first Urban Flood Intelligent Model Python package

pyufim

文档入口

文档	用途
docs/API_REFERENCE.md	独立 API 参考，按 SciPy/xarray 风格列出接口、参数、返回值和注意事项。
docs/PYPI_DEPENDENCIES.md	PyPI 发布依赖清单，包含基础依赖、可选 extra、CUDA 和 native 二进制说明。
docs/qa/GOOGLE_STYLE_AUDIT.md	Google 风格审计结果和存量代码待修复项。
docs/governance/CODE_STYLE.md	后续新增代码必须遵守的中文 Google 风格注释规范。
docs/API.md	既有功能说明和示例。
docs/ITERATION_LOG_100.md	本轮融合 master prompt 的 100 个自检迭代点。

2026-06-18 业务层验证补充

• pyufim-work 已新增 8 个基于 pyufim 的 UFIM 官网样例业务脚本，覆盖单独一维、单独二维、一二维松耦合、一二维紧耦合，以及生成 INP/导入 INP 两类输入方式。
• 修复 add_swmm_outfall(..., outfall_type="FREE") 的 INP 行生成方式，避免 UFIM 1D 内核把 * NO 解析为无效对象。

2026-06-18 最新接口文档和兼容性结论

完整接口文档见 docs/API.md。50 轮自检迭代日志见 docs/ITERATION_LOG.md。

本次迭代后，包内 bin/ 和 src/pyufim/bin/ 同时包含 Windows DLL 与 Linux SO，load_native() 会按当前系统自动选择。

native 库	Windows	Linux	作用
ufimcatchment	ufimcatchment.dll	ufimcatchment.so	D8/DINF 集水区、流向、累积。
ufimdata	ufimdata.dll	ufimdata.so / libufimdata.so	降水和数据清洗校正。
ufimhydro	ufimhydro.dll	ufimhydro.so / libufimhydro.so	RFSM、局部惯性、浅水通量、稳定时间步。
ufimwq	ufimwq.dll	ufimwq.so / libufimwq.so	污染负荷、二维平流扩散、源贡献、溯源。

Windows LISFLOOD/UFIM 2D 的 ufim2d.dll 依赖 gdal309.dll 和 netcdf.dll。本地 shiyao 环境只有 gdal.dll 文件名，因此包内 bin/ 和 src/pyufim/bin/ 同步放置 gdal309.dll 适配副本；加载过程只使用进程级 DLL 搜索目录，不修改系统或 conda 环境变量。

三环境测试结果：

环境	结果
Windows 本机 conda activate shiyao	39 passed；CUDA 不可用时 CUDA 条件测试跳过；DLL 可加载。
Ubuntu CPU 192.168.0.162:2222 /root/miniconda3/envs/shiyao	39 passed, 1 warning；SO 可加载；CUDA 不可用。
Ubuntu CUDA 192.168.0.163:2222 /root/miniconda3/envs/shiyao	40 passed, 1 skipped, 1 warning；CuPy 14.1.1；4 张 RTX 4090 可见；CUDA 水质和通量路径通过。

新增/更新接口重点：

类别	接口
后端选择	BackendInfo、BackendName、is_cuda_available、normalize_backend、resolve_backend，常用计算函数支持 backend="auto/native/numpy/cuda"。
水动力	rfsm_fill_spill、local_inertial_step、local_inertial_timestep、diffusive_wave_timestep、shallow_water_flux_x、water_volume、mass_balance_error。
水质	WaterQualityEngine、pollutant_load、advection_diffusion_2d、source_contribution、trace_sources_2d，支持 native 和 CUDA。
二维任务	TwoDOptions.backend、run_2d() 支持 SCS、RFSM/NATIVE_RFSM、LOCAL_INERTIAL/NATIVE_LOCAL_INERTIAL。
SWMM	新增节点、出流口、调蓄池、管道、断面、泵、堰、孔口、雨量站、时间序列、动力波配置辅助函数。
可视化	extract_raster_profile、water_surface_profile、plot_longitudinal_profile，用于科研纵剖面图。

已知限制：当前 CUDA 已覆盖水质扩散和部分水动力向量算子，完整二维浅水方程整步 GPU 内核、超大 DEM 端到端分块 IO、紧耦合一二维和二维水质耦合还需要继续迭代。

pyufim 是面向城市暴雨内涝建模的 Python 包。它把业务代码需要反复写的“数据检查、降雨与下渗计算、DEM 集水区划分、SWMM 一维模型、二维模型、耦合、结果统计、参数率定入口、命令行和 XML 工作流”统一成可导入的 Python 接口；耗时核心算法通过 bin/ 下的 DLL/SO 调用，不把 C/C++ 源码直接放进 Python 包目录。

当前版本号：1.0.1

功能概览

功能域	能力	主要入口
数据检查	自动识别 CSV、ASC、NetCDF、GRIB、MICAPS、栅格、矢量数据并返回元信息	pyufim.data.inspect_dataset
降水导入	单文件/多文件/目录导入雨量和雨强，支持 CSV、NetCDF、GRIB、MICAPS 000/DAT，自动清洗、校正、时间空间裁剪	pyufim.data.load_precipitation
数据准备	用户输入时间段和空间范围后，自动导入、清洗、裁剪并写出中间数据；当前支持降水和 ASC 栅格	pyufim.data.prepare_dataset
GIS 基础	ASC 读写、坐标与栅格行列转换、最近邻重采样	pyufim.services.gis
水文栅格	D8/DINF 流向、汇流累积、按阈值划分子集水区，支持 ASC 和 NetCDF DEM	pyufim.gis
降雨	外部降水导入、Chicago 雨型、IDF 公式、设计雨型、站点 IDW、LISFLOOD 降雨文件	pyufim.rainfall
下渗	Horton 下渗、SCS-CN 产流	pyufim.infiltration
水文模型	推理公式、单位线、SCS 单位线、时间-面积法、线性水库、Green-Ampt、新安江简化产流	pyufim.hydrology
水利构筑物	堰、孔口、闸门、泵、涵洞、调蓄池连续方程	pyufim.hydraulics
SWMM/一维	INP 读写、对象参数修改、SWMM DLL 运行、逐步运行、结果读取	pyufim.swmm、pyufim.core.SWMMEngine
SWMM 功能配置	暴露 SWMM 下渗、汇流、水质、污染物、土地利用、积累/冲刷、LID、处理表达式等 INP 配置	pyufim.swmm.features
二维	LISFLOOD 原生接口、SCS 面产流接口、RFSM 轻量蓄满扩散、扩散波/局部惯性时间步、浅水通量	pyufim.hydrodynamics、pyufim.models
一二维耦合	管网节点映射到二维网格、堰/孔口交换流量、向 INP 注入入流	pyufim.services.coupling
LID	透水铺装、生物滞留/雨水花园、绿色屋顶、调蓄单元简化水量平衡	pyufim.lid
水质	污染负荷、累积冲刷、一级衰减、二维平流扩散、水质溯源贡献率	pyufim.water_quality
外部模型	Triton、MIKE、ICM 等外部程序命令封装	pyufim.core.external、pyufim.services.external_engines
结果分析	最大水深、到达时间、淹没历时、统计评分	pyufim.results、pyufim.services.stats
可视化	曲线、栅格 PNG、GIF 动图、MP4/MOV/AVI 视频	pyufim.visualization
参数率定	参数定义、RMSE/MAE/Bias/NSE 指标、率定/敏感性/不确定性预留入口	pyufim.calibration
项目与工作流	UFIM 项目目录、UFZ 打包/解包、XML 工作流执行	pyufim.project、pyufim.workflow
命令行	数据检查、通用数据准备、降水裁剪、集水区划分、XML 工作流运行	pyufim

安装

正式 PyPI 安装：

python -m pip install pyufim==1.0.1

开发安装：

cd D:\document\work\codes\pyufim\pyufim-extention
python -m pip install -e .

基础依赖只有 numpy 和 pandas。可选能力按需安装：

python -m pip install -e ".[gis]"
python -m pip install -e ".[meteo]"
python -m pip install -e ".[swmm]"
python -m pip install -e ".[plot]"
python -m pip install -e ".[calibration]"
python -m pip install -e ".[dev]"

可选依赖含义：

extra	依赖	用途
gis	rasterio、geopandas、shapely、pyproj、xarray、netCDF4	栅格、矢量、NetCDF 读取和检查
meteo	xarray、netCDF4、h5py、cfgrib	NetCDF/GRIB 气象网格数据
swmm	swmm-toolkit、pyswmm	后续 SWMM 生态集成
plot	matplotlib、pillow、imageio	曲线、栅格图和动画
calibration	scipy、optuna、scikit-learn	参数率定、敏感性、不确定性分析
dev	pytest、ruff、mypy	测试、格式检查、类型检查

快速示例

自动导入、校正并裁剪降水数据

用户可以直接给原始降水文件和时间/空间范围，不需要先手工裁剪、清洗或转换格式。

from pyufim.data import load_precipitation, save_precipitation_npz

rain = load_precipitation(
    ["rain_20260618.000", "rain_20260619.nc"],
    kind="depth",
    unit="mm",
    variable="PRE",
    time_range=("2026-06-18 00:00", "2026-06-19 23:00"),
    bounds=(119.0, 29.0, 121.0, 31.0),
    correction={"scale": 1.05, "offset": 0.0, "max_value": 300.0},
    clean=True,
    allow_errors=True,
)

save_precipitation_npz(rain, "runs/meteo/rain_clean.npz")

命令行也可以直接做同样的处理：

pyufim precip rain_20260618.000 rain_20260619.nc --kind depth --unit mm --variable PRE --time-start "2026-06-18 00:00" --time-end "2026-06-19 23:00" --bbox 119 29 121 31 --out-npz runs\meteo\rain_clean.npz --json

通用数据准备

统一入口适合业务代码和 XML 工作流复用：同一个函数处理降水、ASC 栅格，后续可以继续扩展矢量、通用 NetCDF 等。

from pyufim.data import DataPrepareRequest, prepare_dataset

prepared = prepare_dataset(
    DataPrepareRequest(
        source="dem.asc",
        kind="asc_raster",
        bounds=(119.0, 29.0, 121.0, 31.0),
        output_path="runs/data/dem_crop.asc",
    )
)

print(prepared.kind)
print(prepared.metadata)

命令行：

pyufim prepare dem.asc --kind asc_raster --bbox 119 29 121 31 --output runs\data\dem_crop.asc --json

基于 ASC DEM 划分 D8 子集水区

from pyufim.gis import delineate_watersheds

result = delineate_watersheds(
    "dem.asc",
    threshold=1000,
    algorithm="D8",
    output_prefix="runs/catchment/demo",
)

print(result.algorithm)
print(result.dem_shape)
print(result.labels)

输出文件：

文件	含义
demo_d8_dir.asc	D8 流向编号，0 表示无有效下游
demo_acc.asc	汇流累积量
demo_subcatch.asc	子集水区标签，0 表示 nodata

基于 NetCDF DEM 划分 DINF 子集水区

from pyufim.gis import delineate_watersheds_from_nc

result = delineate_watersheds_from_nc(
    "dem.nc",
    variable="elevation",
    threshold=2000,
    algorithm="DINF",
    output_path="runs/catchment/watersheds.nc",
)

输出 NetCDF 会写入：

变量	含义
accumulation	汇流累积
watershed_label	子集水区标签

运行 SCS 二维面产流

from pyufim.models import build_2d_model, run_2d

model = build_2d_model(
    "dem.asc",
    options={"model": "SCS"},
    rainfall_mm=80.0,
    curve_number=82.0,
)

summary = run_2d(model, "runs/two_d")
print(summary["runoff_depth_m"])

读取和修改 SWMM INP

from pyufim.swmm import read_inp, write_inp

doc = read_inp("model.inp")
doc.set_option("FLOW_UNITS", "CMS")
doc.set_time_window(start, end)
doc.inject_inflows({"J1": 0.2, "J2": 0.5})
write_inp(doc, "model_modified.inp")

XML 工作流

pyufim 包内不强制用户使用 XML。需要 XML 配置时，建议在业务层，例如 pyufim-work，写 XML 并调用 pyufim.workflow.run_workflow_xml。

<pyufim_workflow>
  <catchment algorithm="D8">
    <dem>../../hbtest/dem12.5m.asc</dem>
    <threshold>1000</threshold>
    <output_prefix>../runs/catchment/dem12_5m</output_prefix>
  </catchment>

  <two_d model="SCS">
    <dem>../../hbtest/dem12.5m.asc</dem>
    <rainfall_mm>80</rainfall_mm>
    <curve_number>82</curve_number>
    <output_dir>../runs/two_d/scs</output_dir>
  </two_d>
</pyufim_workflow>

from pyufim.workflow import run_workflow_xml

summary = run_workflow_xml("config/pyufim_workflow.xml")

D8/DINF 语义

参数/结果	说明
algorithm="D8"	单流向，按最陡下降邻元建图
algorithm="DINF"	双接收者连续流向近似，累积量按比例分配，标签归属使用主流向
threshold	每个子集水区允许的最大有效格子数量
labels	uint32 标签数组，所有有效 DEM 格子均分配正标签，nodata 为 0
use_native=True	优先调用 bin/ufimcatchment.dll 或 Linux 下的 ufimcatchment.so
use_native=False	使用 Python fallback，仅允许 200 万格以内，主要用于测试和算法对照

注意：当前 fill_sinks 是稳定 API，但原生填洼和去平还未完成。若要与 ArcGIS、pysheds 在原始 DEM 上逐格严格对齐，应该先完成 fill depressions 和 resolve flats 的高性能原生实现。

顶层公开接口

这些对象可直接从 pyufim 导入：

from pyufim import (
    __version__,
    ConfigError,
    TimeWindow,
    load_xml,
    make_time_index,
    parse_timedelta,
    CatchmentEngine,
    CatchmentError,
    DinfDirection,
    compute_flow_direction,
    compute_flow_accumulation,
    delineate_watersheds,
    delineate_watersheds_from_nc,
    SWMMEngine,
    SWMMObject,
    SWMMProperty,
    SWMMNodeType,
    SWMMLinkType,
    SWMMFlowUnits,
    SimulationResult,
    LisfloodEngine,
    TwoDParameters,
    ExternalRunSpec,
    ExternalRunResult,
    NativeExternalRunner,
    ExternalEngineAdapter,
    TritonAdapter,
    MikeAdapter,
    IcmAdapter,
    command_to_native_line,
    UnifiedEngineConfig,
    UnifiedEngineResult,
    UnifiedEngine,
    SWMMUnifiedEngine,
    LisfloodUnifiedEngine,
    ExternalUnifiedEngine,
    make_unified_engine,
    load_engine_configs,
    run_engine_config_xml,
    EngineKind,
    INPDocument,
    InpError,
    inp_from_xml,
    PrecipitationData,
    PrecipitationTimeRange,
    SpatialBounds,
    load_precipitation,
    read_precipitation_file,
    save_precipitation_npz,
    list_model_choices,
    UfimProject,
    load_native,
    require_native,
    is_native_available,
    get_native_version,
)

API 文档

聚合模块

pyufim.core 重新导出低层原生封装和外部模型适配器：

from pyufim.core import (
    CatchmentEngine,
    CatchmentError,
    DinfDirection,
    SWMMEngine,
    SWMMObject,
    SWMMProperty,
    SWMMNodeType,
    SWMMLinkType,
    SWMMFlowUnits,
    SimulationResult,
    LisfloodEngine,
    TwoDParameters,
    ExternalRunSpec,
    ExternalRunResult,
    NativeExternalRunner,
    ExternalEngineAdapter,
    TritonAdapter,
    MikeAdapter,
    IcmAdapter,
    command_to_native_line,
)

pyufim.services 重新导出业务服务层常用对象：

from pyufim.services import (
    AscGrid,
    read_asc,
    write_asc,
    point_to_cell,
    cell_to_xy,
    ChicagoRainfallParams,
    chicago_rainfall,
    write_lisflood_rain,
    HortonParams,
    horton_series,
    horton_step,
    CouplingNodeColumns,
    map_nodes_to_grid,
    compute_weir_orifice_exchange,
    inject_inflows_to_inp,
    LisfloodComponentService,
    LisfloodOutputFiles,
    write_bci,
    write_bdy,
    OneDService,
    TwoDService,
    SWMMModelService,
    SWMMReportTables,
    make_adapter,
    run_external_model,
    score_series,
    summarize_frame,
    UnifiedEngineConfig,
    UnifiedEngineResult,
    UnifiedEngine,
    SWMMUnifiedEngine,
    LisfloodUnifiedEngine,
    ExternalUnifiedEngine,
    make_unified_engine,
    load_engine_configs,
    run_engine_config_xml,
)

pyufim.gis

高级水文栅格接口，推荐业务代码优先使用。

class CatchmentResult:
    algorithm: str
    direction: np.ndarray | DinfDirection
    accumulation: np.ndarray
    labels: np.ndarray
    dem_shape: tuple[int, int]
    threshold: int

接口	用途
fill_sinks(dem, nodata=-9999.0) -> np.ndarray	DEM 填洼预处理入口；当前返回拷贝，后续接原生填洼
compute_flow_direction(dem, *, algorithm="D8", nodata=-9999.0, engine=None, use_native=True) -> np.ndarray | DinfDirection	计算 D8 或 DINF 流向
compute_flow_accumulation(direction, *, algorithm="D8", cell_area=1.0, engine=None, use_native=True) -> np.ndarray	计算汇流累积
delineate_watersheds(dem, threshold, *, algorithm="D8", nodata=None, cell_area=None, output_prefix=None, engine=None, use_native=True, progress=None) -> CatchmentResult	从 ASC 或二维数组划分子集水区
delineate_watersheds_from_nc(dem_nc, *, variable=None, threshold, algorithm="D8", nodata=-9999.0, output_path=None, engine=None, use_native=True) -> CatchmentResult	从 NetCDF DEM 划分子集水区

pyufim.core.catchment

原生集水区 DLL/SO 的低层封装。

class DinfDirection:
    dir1: np.ndarray
    dir2: np.ndarray
    frac1: np.ndarray

class CatchmentEngine:
    version() -> str
    d8_flow_direction(dem, nodata=-9999.0) -> np.ndarray
    d8_accumulation(direction, cell_area=1.0) -> np.ndarray
    d8_subcatchments(direction, accumulation, threshold) -> np.ndarray
    dinf_flow_direction(dem, nodata=-9999.0) -> DinfDirection
    dinf_accumulation(direction, cell_area=1.0) -> np.ndarray
    dinf_subcatchments(direction, accumulation, threshold) -> np.ndarray

类/异常	用途
CatchmentEngine	加载并调用 ufimcatchment 原生库
DinfDirection	DINF 双接收者流向结果
CatchmentError	原生库调用失败时抛出

pyufim.services.catchment

兼容旧业务服务层的集水区入口。

接口	用途
delineate_d8(dem, threshold, nodata=-9999.0, cell_area=1.0, engine=None) -> CatchmentResult	二维数组 D8 划分
delineate_dinf(dem, threshold, nodata=-9999.0, cell_area=1.0, engine=None) -> CatchmentResult	二维数组 DINF 划分
delineate_asc(dem_path, threshold, output_prefix, algorithm="D8", nodata=None, engine=None) -> CatchmentResult	ASC 文件划分并写出结果

pyufim.services.gis

ASC 栅格基础工具。

class AscGrid:
    header: dict[str, float]
    data: np.ndarray
    ncols() -> int
    nrows() -> int
    xllcorner() -> float
    yllcorner() -> float
    cellsize() -> float
    nodata() -> float

接口	用途
read_asc(path, dtype=float) -> AscGrid	读取 ESRI ASCII Grid
write_asc(path, grid, fmt="%.6f") -> Path	写出 ESRI ASCII Grid
point_to_cell(grid, x, y) -> tuple[int, int]	坐标转行列号
cell_to_xy(grid, row, col, center=True) -> tuple[float, float]	行列号转坐标
resample_nearest(source, target_header) -> AscGrid	最近邻重采样到目标网格
crop_asc(grid, bounds, *, pad_cells=0) -> AscGrid	按 bbox 裁剪 ASC 栅格

pyufim.data

数据元信息检查、降水导入、自动清洗校正和时空裁剪。

class SpatialBounds:
    min_lon: float
    min_lat: float
    max_lon: float
    max_lat: float
    contains(lon, lat) -> np.ndarray

class TimeRange:
    start: pandas.Timestamp | None
    end: pandas.Timestamp | None
    mask(times) -> np.ndarray

class PrecipitationData:
    values: np.ndarray
    dims: tuple[str, ...]
    unit: Literal["mm", "m", "mm/h", "m/s"]
    kind: Literal["depth", "intensity"]
    times: pd.DatetimeIndex | None
    lon: np.ndarray | None
    lat: np.ndarray | None
    source_files: tuple[Path, ...]
    metadata: dict[str, Any]
    warnings: tuple[str, ...]
    crop(time_range=None, bounds=None) -> PrecipitationData
    clean(...) -> PrecipitationData
    apply_correction(correction=None, **kwargs) -> PrecipitationData
    to_unit(target_unit, interval_seconds=None) -> PrecipitationData
    to_dataframe() -> pd.DataFrame

class DataPrepareRequest:
    source: str | Path | Sequence[str | Path]
    kind: Literal["auto", "precipitation", "asc_raster"]
    output_path: str | Path | None
    time_range: Sequence[Any] | None
    bounds: Sequence[float] | None
    variable: str | None
    precipitation_kind: Literal["auto", "depth", "intensity"]
    precipitation_unit: Literal["mm", "m", "mm/h", "m/s"] | None
    correction: Mapping[str, Any] | None
    clean: bool
    allow_errors: bool
    pad_cells: int

class PreparedData:
    kind: Literal["precipitation", "asc_raster", "unknown"]
    source: tuple[Path, ...]
    output_path: Path | None
    data: PrecipitationData | AscGrid | None
    metadata: dict[str, Any]
    warnings: tuple[str, ...]

接口	用途
inspect_dataset(path) -> dict[str, Any]	根据扩展名自动检查数据
inspect_raster(path) -> dict[str, Any]	检查栅格或 ASC 文件
inspect_vector(path) -> dict[str, Any]	检查矢量文件，需要 geopandas
inspect_timeseries(path, time_col=None, value_col=None) -> dict[str, Any]	检查 CSV 时间序列
load_precipitation(paths, *, kind="auto", unit=None, source_format="auto", variable=None, time_range=None, bounds=None, correction=None, clean=True, allow_errors=True, encoding=None, time_col=None, lon_col=None, lat_col=None, value_col=None, micaps_time=None) -> PrecipitationData	自动导入单个文件、多个文件或目录中的降水数据
read_precipitation_file(path, *, kind="auto", unit=None, source_format="auto", variable=None, encoding=None, time_col=None, lon_col=None, lat_col=None, value_col=None, micaps_time=None) -> PrecipitationData	读取单个降水文件
detect_precipitation_format(path) -> Literal["table", "netcdf", "grib", "micaps"]	识别降水格式
inspect_precipitation_file(path) -> dict[str, Any]	快速检查降水文件格式和头信息
combine_precipitation(datasets, *, allow_mismatch=True) -> PrecipitationData	合并多个降水数据
crop_precipitation(data, *, time_range=None, bounds=None) -> PrecipitationData	按时间和空间裁剪
apply_precipitation_correction(data, *, scale=1.0, offset=0.0, min_value=0.0, max_value=None, fill_value=0.0, missing_values=(-9999, 999, 9999, 32766, 65535), clip_negative=True, custom=None) -> PrecipitationData	规则化降水校正
save_precipitation_npz(data, output_path) -> Path	保存清洗裁剪后的统一降水中间文件
prepare_dataset(request) -> PreparedData	统一数据准备入口，当前支持降水和 ASC 栅格
prepare_many(requests) -> list[PreparedData]	批量准备多个数据集

降水格式支持：

格式	扩展名	依赖	说明
表格	.csv、.txt、.tsv	基础依赖	支持时间序列、站点长表
NetCDF	.nc	pyufim[meteo]	自动识别 time/lon/lat 坐标，可指定变量
GRIB	.grib、.grb、.grb2	pyufim[meteo] 和 ecCodes 环境	通过 xarray/cfgrib 读取
MICAPS	.000、.dat	基础依赖	支持常见 MICAPS 网格和站点文本

pyufim.rainfall

降雨计算和文件生成。

class ChicagoRainfallParams:
    a: float
    c: float
    return_period: int
    b: float
    n: float
    peak_ratio: float

class IDFFormula:
    name: str
    expression: str
    parameters: Mapping[str, float]
    time_unit: str
    intensity_unit: str
    region: str | None

接口	用途
chicago_rainfall(time_index, params, output_unit="m/s") -> pd.Series	生成 Chicago 雨型
write_lisflood_rain(path, rainfall, times=None, input_unit="m/s") -> Path	写 LISFLOOD 降雨文件
rainfall_to_grid(value, shape) -> np.ndarray	常数降雨转二维栅格
clean_rainfall(data, unit) -> np.ndarray	降雨单位和无效值清洗
station_idw(stations, lon, lat, power=1.0, value_col="value") -> np.ndarray	站点降雨 IDW 插值
register_idf_formula(formula) -> None	注册 IDF 公式
load_idf_formula(path) -> IDFFormula	从 JSON 读取 IDF 公式
compute_rainfall_intensity(formula, duration_minutes, return_period_year) -> float	计算暴雨强度
generate_design_rainfall(formula, *, start, duration, interval="5min", return_period_year=20, peak_ratio=0.4) -> pd.Series	生成设计雨型序列
load_precipitation(...) -> PrecipitationData	从降雨模块直接导入外部降水数据
read_precipitation_file(...) -> PrecipitationData	读取单个外部降水文件
save_precipitation_npz(data, output_path) -> Path	保存降水中间文件

pyufim.infiltration

下渗和 SCS 产流。

class HortonParams:
    f0_m_s: float
    fmin_m_s: float
    decay_1_s: float
    recover_1_s: float
    max_volume_m: float

接口	用途
horton_series(rainfall_depth_m, delta_seconds, params, initial_capacity_m_s=None, initial_cumulative_m=0.0) -> dict[str, np.ndarray]	计算 Horton 下渗时间序列
horton_step(capacity_m_s, available_water_m, cumulative_m, delta_seconds, params) -> tuple[np.ndarray, np.ndarray, np.ndarray]	单步 Horton 下渗
scs_runoff_depth(rainfall_mm, curve_number, initial_abstraction_ratio=0.2) -> np.ndarray	SCS-CN 产流深，单位米

pyufim.swmm

SWMM 业务友好入口。

接口	用途
read_inp(path) -> INPDocument	读取 SWMM INP
write_inp(document, path) -> Path	写出 SWMM INP
validate_inp_syntax(path) -> dict[str, object]	基础语法检查
validate_inp_with_swmm(path) -> dict[str, object]	调 SWMM 引擎校验
run_swmm(inp_path, output_dir, *, prefix="report")	运行 SWMM 并写 report/out
parse_rpt(path) -> str	读取 RPT 文本
set_swmm_infiltration_model(document, model) -> INPDocument	设置 SWMM 下渗模型，如 Horton、Green-Ampt、SCS-CN
set_swmm_flow_routing(document, routing) -> INPDocument	设置 SWMM 水力演算，如 Steady、Kinematic Wave、Dynamic Wave
add_swmm_pollutant(document, name, units="MG/L", rain_conc=0.0, gw_conc=0.0, init_conc=0.0, decay=0.0) -> INPDocument	添加污染物
add_swmm_landuse(document, name, street_sweeping_interval_days=0.0, availability=0.0, last_swept_days=0.0) -> INPDocument	添加土地利用
add_swmm_buildup(document, landuse, pollutant, function, coefficient1, coefficient2, coefficient3=0.0, normalizer="AREA") -> INPDocument	添加污染累积函数
add_swmm_washoff(document, landuse, pollutant, function, coefficient1, coefficient2, cleaning_efficiency=0.0, bmp_efficiency=0.0) -> INPDocument	添加污染冲刷函数
add_swmm_lid_control(document, name, lid_type, layers) -> INPDocument	添加 SWMM LID 控制
add_swmm_lid_usage(document, subcatchment, lid_control, number, area_m2, width_m, ...) -> INPDocument	添加 SWMM LID 使用
add_swmm_treatment(document, node_or_link, pollutant, expression) -> INPDocument	添加水质处理表达式
configure_swmm_quality(document, *, route_step=None, wet_step=None, report_step=None) -> INPDocument	配置 SWMM 水质相关时间步

pyufim.inp

SWMM INP 文本结构化编辑。

class INPDocument:
    read(path, encoding="utf-8") -> INPDocument
    from_sections(sections) -> INPDocument
    write(path, encoding="utf-8") -> Path
    to_text() -> str
    get_section(section) -> list[str]
    append_line(section, line) -> None
    set_option(option, value) -> None
    set_time_window(start, end) -> None
    set_hotstart_files(save_hotstart=None, use_hotstart=None, replace_existing=True) -> None
    find_object_line(section, object_id) -> int
    set_object_token(section, object_id, token_index, value) -> None
    inject_inflows(inflows, overwrite=True, pollutant_type="FLOW") -> None

接口	用途
INPDocument	按 section 操作 INP 文本
inp_from_xml(xml_path, output_path) -> Path	从 XML 生成 INP 的兼容入口
InpError	INP 解析或修改错误

pyufim.core.swmm

SWMM DLL 低层封装。

class SimulationResult:
    code: int
    report_path: Path | None
    output_path: Path | None
    message: str
    ok() -> bool

class SWMMEngine:
    run(inp_path, report_path, output_path, check=True) -> SimulationResult
    run_stepwise(inp_path, report_path, output_path, save_results=True, stride_step=None) -> list[float]
    open(inp_path, report_path, output_path) -> int
    start(save_results=True) -> int
    step() -> tuple[int, float]
    stride(stride_step) -> tuple[int, float]
    end() -> int
    report() -> int
    close() -> int
    get_count(object_type) -> int
    object_names(object_type) -> list[str]
    get_name(object_type, index, size=128) -> str
    get_index(object_type, name) -> int
    require_index(object_type, name) -> int
    get_value(property_id, index) -> float
    get_value_by_name(object_type, property_id, name) -> float
    values_by_name(object_type, property_id) -> dict[str, float]
    set_value(property_id, index, value) -> None
    set_value_by_name(object_type, property_id, name, value) -> None
    get_saved_value(property_id, index, period) -> float
    saved_series(object_type, property_id, name, periods) -> list[float]
    get_mass_balance_error() -> dict[str, float]
    get_version() -> int
    get_warnings() -> int
    get_error_message(size=1024) -> str
    write_line(line) -> None
    decode_date(date_value) -> dict[str, int]
    merge_coordinates_from_inp_and_csv(inp_path, csv_path) -> None

枚举	用途
SWMMObject	SWMM 对象类型
SWMMProperty	SWMM 属性编号
SWMMNodeType	节点类型
SWMMLinkType	管线类型
SWMMFlowUnits	流量单位

pyufim.services.swmm

SWMM 服务层。

class SWMMReportTables:
    summary: pd.DataFrame | None
    nodes: pd.DataFrame | None
    links: pd.DataFrame | None
    subcatchments: pd.DataFrame | None

class SWMMModelService:
    run(inp_path, report_path, output_path) -> SimulationResult
    open(inp_path, report_path, output_path, save_results=True) -> None
    close(write_report=True) -> None
    step() -> float
    stride(stride_step) -> float
    list_objects(object_type) -> list[str]
    get(object_type, object_name, property_name) -> float
    set(object_type, object_name, property_name, value) -> None
    get_many(object_type, property_name) -> dict[str, float]
    saved_series(object_type, object_name, property_name, periods) -> pd.Series
    mass_balance_error() -> dict[str, float]
    version() -> int
    warnings() -> int
    read_report_tables(output_dir, prefix="report") -> SWMMReportTables
    concat_reports(output_dirs, prefix="report") -> SWMMReportTables

pyufim.core.lisflood

LISFLOOD 二维模型低层封装。

class TwoDParameters:
    dem_file: str | Path
    res_root: str
    dir_root: str | Path
    sim_time: int | str
    initial_tstep: int | str
    massint: int | str
    saveint: int | str
    overpass: int | str
    fpfric: float | str
    bci_file: str | Path | None
    bdy_file: str | Path | None
    rainfall_file: str | Path | None
    start_file: str | Path | None
    extra_lines: tuple[str, ...]

class LisfloodEngine:
    has_function(name) -> bool
    create_bci(csv_path, bci_path) -> Path
    create_bdy(csv_path, bdy_path) -> Path
    create_par(par_path, params, use_native_writer=True) -> Path
    run(par_path, check=True) -> int
    convert_wd_to_tif(wd_path, tif_path) -> Path
    convert_wd_to_geojson(wd_path, geojson_path) -> Path

接口	用途
write_par_file(path, params) -> Path	Python 侧写 LISFLOOD .par 文件
TwoDParameters	二维模型参数
LisfloodEngine	调用 ufim2d 原生库

pyufim.services.lisflood

LISFLOOD 业务组件。

class LisfloodOutputFiles:
    wd: list[Path]
    elev: list[Path]
    qx: list[Path]
    qy: list[Path]
    max_depth: list[Path]

class LisfloodComponentService:
    create_bci(boundary_csv, bci_path) -> Path
    create_bdy(boundary_csv, bdy_path) -> Path
    write_par(path, params) -> Path
    run_par(par_path) -> int
    convert_wd(wd_path, tif_path=None, geojson_path=None) -> dict[str, Path]
    convert_all_wd(directory, tif=True, geojson=True) -> list[dict[str, Path]]
    collect_outputs(directory) -> LisfloodOutputFiles
    latest_wd(directory) -> Path | None

接口	用途
write_bci(boundary_csv, bci_path) -> Path	写 BCI 边界文件
write_bdy(boundary_csv, bdy_path, value_col=None) -> Path	写 BDY 边界文件
write_uniform_boundary_csv(path, rows) -> Path	写统一边界 CSV

pyufim.models

模型编排层，业务代码优先使用这里选择 1D/2D/耦合模型。

class OneDOptions:
    flow_units: Literal["CMS", "LPS", "MLD"]
    routing_method: Literal["STEADY", "KINWAVE", "DYNWAVE"]
    infiltration_model: Literal["HORTON", "MOD_HORTON", "GREEN_AMPT", "MOD_GREEN_AMPT", "SCS_CN"]
    allow_ponding: bool
    threads: int | None

class TwoDOptions:
    duration: datetime.timedelta
    dt: datetime.timedelta
    model: Literal["LISFLOOD", "SCS", "RFSM", "CELLULAR", "DIFFUSIVE_WAVE", "LOCAL_INERTIAL", "SHALLOW_WATER"]
    solver: Literal["cellular", "diffusive_wave", "local_inertial", "shallow_water"]
    global_manning_n: float | None
    cfl: float
    output_interval: datetime.timedelta | None
    boundary_condition: dict[str, Any]
    use_landuse_manning: bool
    use_infiltration_grid: bool
    threads: int | None
    tile_size: int | None

class CouplingOptions:
    mode: Literal["loose", "tight", "surface_only", "one_d_only"]
    exchange_dt: datetime.timedelta | None
    report_dt: datetime.timedelta | None
    allow_overflow_to_surface: bool
    allow_surface_return_to_network: bool
    inlet_capacity_model: Literal["unlimited", "orifice", "weir", "rating_curve"]
    mass_balance_tolerance: float

class OneDModel:
    inp_path: Path
    options: OneDOptions

class TwoDModel:
    dem: Path
    rainfall: Path | None
    boundary: Path | None
    landuse: Path | None
    options: TwoDOptions
    parameters: dict[str, Any]

class CoupledModel:
    one_d: OneDModel
    two_d: TwoDModel
    coupling_map: dict[str, Any]
    options: CouplingOptions

可选模型和方法可以通过 list_model_choices() 集中读取，返回字段包括：

字段	说明
one_d_routing_methods	一维汇流方法
one_d_infiltration_models	一维下渗模型
two_d_models	二维模型族，包括 LISFLOOD、SCS、RFSM 等
two_d_solvers	二维求解器
coupling_modes	一二维耦合模式
inlet_capacity_models	雨水口/节点交换能力模型
calibration_methods	参数率定方法
hydrology_models	水文模型
hydraulic_structure_models	水利构筑物模型
water_quality_models	水质模型
lid_models	LID 模型
swmm_quality_sections	SWMM 水质和 LID 相关 INP section

接口	用途
build_2d_model(dem, *, rainfall=None, boundary=None, landuse=None, options=None, **parameters) -> TwoDModel	创建二维模型对象
build_coupling_map(one_d_network=None, two_d_dem=None) -> dict[str, Any]	构建耦合映射入口
list_model_choices() -> dict[str, tuple[str, ...]]	返回一维、二维、耦合、雨水口和率定可选项
run_1d(model_or_inp, output_dir, *, prefix="report", check=True) -> SimulationResult	运行 SWMM 一维
run_2d(model, output_dir, *, check=True) -> dict[str, Any] | int	运行二维模型；SCS 已实现，LISFLOOD 调原生，RFSM 为预留入口
run_loose_coupling(model, output_dir, *, check=True) -> dict[str, Any]	松耦合运行入口
run_tight_coupling(model, output_dir, *, check=True) -> dict[str, Any]	紧耦合预留入口
run_coupled(model, output_dir, *, check=True) -> dict[str, Any]	根据 CouplingOptions.mode 调度耦合运行

pyufim.services.simulation

一维/二维运行服务。

class OneDStepPaths:
    inp: Path
    report: Path
    output: Path
    hotstart: Path

class OneDService:
    run(inp_path, output_dir, prefix="report", check=True) -> SimulationResult
    prepare_step_inp(template_inp, output_dir, start, end, use_hotstart=None, inflows=None) -> OneDStepPaths
    run_step(template_inp, output_dir, start, end, use_hotstart=None, inflows=None, check=True) -> tuple[SimulationResult, OneDStepPaths]

class TwoDService:
    prepare_boundary_run(csv_path, params, output_dir, par_name="run.par") -> Path
    prepare_rainfall_run(params, output_dir, par_name="rainfall.par") -> Path
    run(par_path, check=True) -> int

pyufim.hydrology

轻量水文模型。用于快速产流、汇流、方案比选和模型前处理。

接口	用途
rational_peak_flow(rainfall_intensity_mm_h, area_km2, runoff_coefficient=0.6)	推理公式峰值流量
unit_hydrograph_convolution(effective_rainfall_mm, unit_hydrograph, area_km2=1.0, dt_seconds=3600.0)	单位线卷积
scs_unit_hydrograph(duration_hours, peak_time_hours, peak_factor=2.08, dt_hours=1.0)	简化 SCS 单位线
time_area_hydrograph(effective_rainfall_mm, time_area_weights, area_km2=1.0, dt_seconds=3600.0)	时间-面积法
linear_reservoir_runoff(inflow, storage_coefficient_seconds, dt_seconds)	线性水库汇流
green_ampt_infiltration(rainfall_depth_m, dt_seconds, hydraulic_conductivity_m_s, suction_head_m, delta_theta, initial_cumulative_m=0.0)	Green-Ampt 下渗
xin_an_jiang_runoff(rainfall_mm, evaporation_mm=0.0, soil_storage_mm=80.0, tension_water_capacity_mm=100.0, runoff_exponent=1.5)	简化新安江蓄满产流

pyufim.hydraulics

水利构筑物和调蓄公式。

接口	用途
weir_flow(head_m, width_m, coefficient=1.7, submergence_factor=1.0)	矩形堰流量
orifice_flow(head_m, area_m2, coefficient=0.62, gravity=9.80665)	孔口流量
gate_flow(upstream_head_m, downstream_head_m, opening_m, width_m, coefficient=0.62, gravity=9.80665)	闸门孔流量
pump_flow(head_m, curve, *, shutoff_head_m=None)	泵曲线插值
culvert_flow(headwater_m, tailwater_m, diameter_m, length_m, mannings_n=0.013, inlet_coefficient=0.6, gravity=9.80665)	简化涵洞流量
storage_continuity(storage_m3, inflow_m3_s, outflow_m3_s, dt_seconds, min_storage_m3=0.0, max_storage_m3=None)	调蓄池连续方程

pyufim.hydrodynamics

二维水动力轻量接口。完整高精度模拟仍优先使用 LISFLOOD/native 或后续专门求解器。

接口	用途
rfsm_fill_spill(dem_m, water_depth_m, *, iterations=50, relaxation=0.5, nodata=None)	RFSM 近似蓄满扩散
diffusive_wave_timestep(cellsize_m, depth_m, mannings_n=0.06, slope=0.001, cfl=0.7)	扩散波稳定时间步估算
local_inertial_timestep(cellsize_m, depth_m, gravity=9.80665, cfl=0.7)	局部惯性/浅水波速时间步估算
shallow_water_flux_x(depth_m, velocity_m_s)	浅水方程 x 方向通量

pyufim.water_quality

水质负荷、二维扩散和溯源轻量模型。

接口	用途
pollutant_load(flow_m3_s, concentration_mg_l, dt_seconds=1.0)	由流量和浓度计算污染负荷 kg
first_order_decay(concentration, decay_rate_1_s, dt_seconds)	一级衰减
buildup_washoff_load(area_m2, buildup_kg_m2, runoff_m, washoff_coefficient=1.0, exponent=1.0)	地表累积-冲刷负荷
advection_diffusion_2d(concentration, depth_m, u_m_s, v_m_s, diffusivity_m2_s, dx_m, dy_m, dt_seconds, decay_rate_1_s=0.0, nodata=None)	二维显式平流-扩散-衰减一步更新
source_contribution(source_loads)	污染源总量贡献率
trace_sources_2d(source_concentration_layers, weights=None)	二维逐格源贡献率

pyufim.lid

LID 简化水量平衡模型。SWMM LID 的完整工程配置可用 pyufim.swmm 中的 INP 配置函数写入。

接口	用途
storage_unit(inflow_m, capacity_m, drain_rate_m_s, dt_seconds, initial_storage_m=0.0)	通用调蓄单元
permeable_pavement(rainfall_m, dt_seconds, void_depth_m, infiltration_rate_m_s, initial_storage_m=0.0)	透水铺装
bioretention_cell(rainfall_m, dt_seconds, surface_storage_m, soil_storage_m, infiltration_rate_m_s, underdrain_rate_m_s=0.0)	生物滞留/雨水花园
green_roof(rainfall_m, dt_seconds, substrate_storage_m, evapotranspiration_rate_m_s=0.0, drain_rate_m_s=1e-7)	绿色屋顶

pyufim.services.coupling

一二维耦合计算工具。

class CouplingNodeColumns:
    node_id: str
    x: str
    y: str
    head: str
    depth: str
    overflow: str

接口	用途
map_nodes_to_grid(nodes, grid, columns=CouplingNodeColumns()) -> pd.DataFrame	管网节点映射到二维网格行列
compute_weir_orifice_exchange(wd_depth_m, dem_elevation_m, node_map, node_state, seconds, columns=CouplingNodeColumns(), weir_coefficient=0.5, orifice_coefficient=0.5, width_m=0.8, opening_height_m=2.51, gravity=9.8, remove_from_surface=True) -> tuple[pd.DataFrame, np.ndarray]	计算地表和节点之间的交换流量
compute_exchange_from_files(wd_path, dem_path, node_map, node_state, seconds) -> tuple[pd.DataFrame, np.ndarray]	从文件读取水深/DEM 后计算交换
inject_inflows_to_inp(inp_path, inflows, output_path=None, node_col="NodeID", value_col="inflow") -> Path	把交换入流写入 SWMM INP

pyufim.core.external

外部模型程序运行封装。

class ExternalRunSpec:
    executable: str | Path
    args: tuple[str, ...]
    cwd: str | Path | None
    env: Mapping[str, str] | None
    timeout_seconds: float | None
    stdout_path: str | Path | None
    stderr_path: str | Path | None
    metadata: Mapping[str, object]

class ExternalRunResult:
    engine: str
    returncode: int
    command: tuple[str, ...]
    cwd: Path | None
    stdout_path: Path | None
    stderr_path: Path | None
    metadata: Mapping[str, object]
    ok() -> bool

class NativeExternalRunner:
    version() -> str
    run_command(command_line, cwd=None, stdout_path=None, stderr_path=None, timeout_seconds=None) -> int
    write_text(path, text) -> None

class ExternalEngineAdapter:
    make_spec(model_file, output_dir, extra_args=(), timeout_seconds=None) -> ExternalRunSpec
    run(spec, check=True) -> ExternalRunResult
    write_manifest(path, result) -> Path

接口	用途
TritonAdapter	Triton 外部模型适配器
MikeAdapter	MIKE 外部模型适配器
IcmAdapter	ICM 外部模型适配器
command_to_native_line(command) -> str	命令列表转原生命令行字符串

pyufim.services.external_engines

外部模型服务层。

接口	用途
make_adapter(engine, executable=None) -> ExternalEngineAdapter	根据模型名创建适配器
run_external_model(engine, executable, model_file, output_dir, extra_args=(), cwd=None, timeout_seconds=None, check=True) -> ExternalRunResult	运行外部模型

pyufim.engines

统一模型运行接口，用同一套 XML 或配置描述 SWMM、LISFLOOD、外部模型。

class UnifiedEngineConfig:
    name: str
    kind: EngineKind
    model_file: Path
    output_dir: Path
    executable: Path | None
    args: tuple[str, ...]
    cwd: Path | None
    timeout_seconds: float | None
    artifacts: Mapping[str, Path]
    metadata: Mapping[str, str]

class UnifiedEngineResult:
    name: str
    kind: EngineKind
    returncode: int
    output_dir: Path
    artifacts: Mapping[str, Path]
    native_result: SimulationResult | ExternalRunResult | int | None
    ok() -> bool

class UnifiedEngine:
    run(check=True) -> UnifiedEngineResult

接口	用途
EngineKind	引擎类型字面量
SWMMUnifiedEngine	SWMM 统一引擎
LisfloodUnifiedEngine	LISFLOOD 统一引擎
ExternalUnifiedEngine	外部程序统一引擎
make_unified_engine(config) -> UnifiedEngine	按配置创建统一引擎
load_engine_configs(path) -> list[UnifiedEngineConfig]	从 XML 读取引擎配置
run_engine_config_xml(path, name=None, check=True) -> list[UnifiedEngineResult]	执行 XML 中的一个或多个引擎

pyufim.results

二维结果统计。

接口	用途
list_result_variables(result_or_dir) -> list[str]	列出结果变量或目录中的结果文件
compute_max_depth(depth_series, *, nodata=None) -> np.ndarray	计算最大水深
compute_arrival_time(depth_series, *, threshold=0.05, nodata_value=-1) -> np.ndarray	计算首次超过阈值的时间步
compute_inundation_duration(depth_series, *, threshold=0.05, step_seconds=1.0) -> np.ndarray	计算淹没历时，单位秒

pyufim.services.stats

序列评分和表格统计。

接口	用途
score_series(observed, simulated) -> dict[str, float]	计算序列统计指标
align_on_time(observed, simulated, value_col, tolerance="10min", time_col="TimeSeries") -> pd.DataFrame	按时间对齐观测和模拟
summarize_frame(frame, group_col, value_col) -> pd.DataFrame	分组统计
metrics_table(rows) -> pd.DataFrame	指标行列表转表格

pyufim.calibration

参数率定入口。

class ParameterSpec:
    name: str
    target: str
    lower: float
    upper: float
    initial: float | None
    transform: Literal["linear", "log"]
    group: str | None

接口	用途
rmse(sim, obs) -> float	均方根误差
mae(sim, obs) -> float	平均绝对误差
bias(sim, obs) -> float	偏差
nse(sim, obs) -> float	Nash-Sutcliffe 效率系数
calibrate(model, observations, parameters, *, objective, method="differential_evolution", max_evaluations=100, output_dir=None, minimize=True, seed=0) -> dict[str, Any]	通用参数率定；支持 grid、random、differential_evolution，无 scipy 时自动回退 random
sensitivity_analysis(model, parameters, *, method="random_correlation", n=256, output_dir=None, seed=0) -> dict[str, Any]	轻量随机相关敏感性分析，不依赖 SALib/scipy
uncertainty_ensemble(model, parameter_samples, *, output_dir=None) -> dict[str, Any]	参数集合模拟，返回 mean/min/max/p05/p50/p95

model 可以是函数，也可以是带 run(params) 或 simulate(params) 方法的对象。参数率定输出包含：

字段	说明
best_parameters	最优参数字典
best_score	最优目标函数值
evaluations	已运行模型次数
history	每次评估的参数和分数
fallback	可选依赖不可用时的回退说明

pyufim.native

原生库加载工具。

原生库按当前系统自动选择 .dll、.so 或 .dylib。Windows 下使用进程级 os.add_dll_directory，Linux/macOS 下使用候选绝对路径加载，不修改 PATH、LD_LIBRARY_PATH 等环境变量。

接口	用途
load_native(name="catchment", *, bin_dir=None)	加载原生库，支持 catchment、1d/swmm、2d/lisflood、external
require_native(name="catchment", *, bin_dir=None)	加载失败时抛错
is_native_available(name="catchment", *, bin_dir=None) -> bool	检查原生库是否可用
get_native_version(name="catchment", *, bin_dir=None) -> str	读取原生库版本

支持的 name：

名称	Windows	Linux	用途
catchment	ufimcatchment.dll	ufimcatchment.so	D8/DINF 集水区
swmm	ufim1d.dll / ufim.dll	ufim1d.so	SWMM/一维
lisflood	ufim2d.dll	ufim2d.so	二维
external	ufimexternal.dll	ufimexternal.so	外部程序调用
data	ufimdata.dll	libufimdata.so / ufimdata.so	降水数组清洗、线性校正等数据处理核心

pyufim.core.data

数据处理原生 C 后端。当前用于降水错误值清洗、比例校正和上下限裁剪；不可用时 pyufim.data.apply_precipitation_correction 会自动回退到 NumPy。

class DataEngine:
    version() -> str
    clean_precipitation(values, *, missing_values, fill_value, scale, offset, min_value, max_value, clip_negative) -> np.ndarray

pyufim.project

项目目录和 UFZ 打包。

class UfimProject:
    root: Path
    name: str
    config: dict[str, Any]
    data: dict[str, str]
    runs: dict[str, str]
    create(root, name, *, overwrite=False) -> UfimProject
    open(root) -> UfimProject
    save() -> Path
    export_ufz(output_path) -> Path
    import_ufz(ufz_path, root, *, overwrite=False) -> UfimProject

pyufim.config

XML 和时间配置工具。

class TimeWindow:
    start: datetime.datetime
    end: datetime.datetime
    from_values(start, end, fmt="%Y%m%d%H%M") -> TimeWindow

接口	用途
load_xml(path) -> dict[str, Any]	XML 转字典
element_to_data(element) -> Any	XML 元素转 Python 数据
get_nested(data, keys, default=None) -> Any	读取嵌套字段
parse_datetime(value, fmt="%Y%m%d%H%M") -> datetime	解析时间
parse_timedelta(value) -> timedelta	解析时间步长
make_time_index(start, end, step, fmt="%Y%m%d%H%M") -> list[datetime]	生成时间索引
ConfigError	配置错误

pyufim.workflow

业务 XML 运行器。

接口	用途
run_workflow_xml(path) -> dict[str, Any]	执行 <catchment> 和 <two_d> 工作流节点

支持节点：

节点	字段
<precipitation>	file、path、kind、unit、variable、time_start、time_end、bbox、clean、allow_errors、output_npz
<catchment>	algorithm、dem、threshold、output_prefix、variable、output_nc
<two_d>	model、dem、rainfall、rainfall_mm、curve_number、output_dir

降水节点示例：

<precipitation kind="depth" unit="mm">
  <file>rain_20260618.000</file>
  <file>rain_20260619.nc</file>
  <variable>PRE</variable>
  <time_start>2026-06-18 00:00</time_start>
  <time_end>2026-06-19 23:00</time_end>
  <bbox>119 29 121 31</bbox>
  <clean>true</clean>
  <allow_errors>true</allow_errors>
  <output_npz>runs/meteo/rain_clean.npz</output_npz>
</precipitation>

pyufim.services.conversion

格式转换和文件辅助。

接口	用途
wd_to_asc_grid(path) -> AscGrid	水深文件转 ASC 栅格对象
write_boundary_csv(path, rows) -> Path	写边界 CSV
copy_asc(path, output_path) -> Path	复制 ASC

pyufim.services.plotting 和 pyufim.visualization

绘图、动态图和视频工具，需要 pyufim[plot]。pyufim.visualization 重新导出同一组高层函数。

接口	用途
plot_series(frame, x, y_columns, output_path, title=None, dpi=160) -> Path	绘制时间序列
plot_raster(data, output_path, title=None, dpi=160) -> Path	绘制栅格图
save_raster_png(data, output_path, *, title=None, cmap="Blues", vmin=None, vmax=None, nodata=None, dpi=160) -> Path	保存单张栅格 PNG
save_raster_animation(frames, output_path, *, times=None, fps=6, title=None, cmap="Blues", vmin=None, vmax=None, nodata=None, dpi=120) -> Path	把三维栅格序列保存为 GIF 或 MP4/MOV/AVI
frames_to_video(image_paths, output_path, *, fps=6) -> Path	把已有图片序列合成为 GIF 或视频

pyufim.logging

日志配置。

接口	用途
get_logger(name="pyufim") -> logging.Logger	获取 logger
configure_logging(level="INFO", log_file=None) -> None	配置控制台/文件日志

pyufim.paths

路径和原生库搜索路径。

接口	用途
current_os() -> str	返回 windows、linux 或 macos 等
as_path(value) -> Path	转 Path
ensure_dir(path) -> Path	确保目录存在
add_native_library_dirs(bin_dir=None, lib_dir=None, extra_dirs=()) -> None	添加 DLL/SO 搜索路径
native_library_candidates(name) -> list[Path]	返回候选原生库路径

pyufim.typing

接口	用途
ProgressCallback = Callable[[float, str], None]	进度回调类型，第一参数为 0.0-1.0 进度，第二参数为状态文本

pyufim.exceptions

统一异常类型。

异常	用途
UfimError	pyufim 基类异常
DataImportError	数据导入错误
CRSValidationError	坐标系验证错误
SchemaValidationError	数据结构验证错误
SwmmError	SWMM 错误
NativeBackendError	原生后端错误
SimulationError	模型运行错误
CalibrationError	率定错误

命令行接口

安装后命令为 pyufim。

pyufim --help

数据检查

pyufim info path\to\data.asc --json

通用数据准备

pyufim prepare dem.asc --kind asc_raster --bbox 119 29 121 31 --output runs\data\dem_crop.asc --json
pyufim prepare rain1.000 rain2.nc --kind precipitation --precip-kind depth --unit mm --variable PRE --time-start "2026-06-18 00:00" --time-end "2026-06-19 23:00" --bbox 119 29 121 31 --output runs\meteo\rain_clean.npz --json

参数：

参数	说明
paths	一个或多个数据文件，或目录
--kind	auto、precipitation 或 asc_raster
--output	输出路径；降水建议 .npz，ASC 建议 .asc
--time-start / --time-end	模拟时间段裁剪，ASC 会忽略时间并返回 warning
--bbox	空间范围，顺序为 min_x min_y max_x max_y
--variable	NetCDF/GRIB 变量名
--precip-kind	auto、depth 或 intensity
--unit	降水单位
--no-clean	降水不做基础清洗
--strict	多文件导入时任一文件失败即报错
--pad-cells	ASC 裁剪时四周额外保留格子数

集水区划分

pyufim catchment path\to\dem.asc --threshold 1000 --algorithm D8 --out-prefix runs\catchment\demo --json
pyufim catchment path\to\dem.asc --threshold 1000 --algorithm DINF --python-fallback

参数：

参数	说明
dem	DEM 文件，当前 CLI 主要面向 ASC
--threshold	子集水区最大格子数，必填
--algorithm	D8 或 DINF
--out-prefix	输出文件前缀
--python-fallback	不调用原生库，用 Python 小数据 fallback
--json	JSON 输出摘要

降水自动导入和裁剪

pyufim precip rain1.000 rain2.nc --kind depth --unit mm --variable PRE --time-start "2026-06-18 00:00" --time-end "2026-06-19 23:00" --bbox 119 29 121 31 --out-npz runs\meteo\rain_clean.npz --json

参数：

参数	说明
paths	一个或多个降水文件，也可以是目录
--kind	auto、depth 或 intensity
--unit	mm、m、mm/h 或 m/s
--variable	NetCDF/GRIB 变量名
--time-start / --time-end	模拟时间段裁剪
--bbox	空间范围，顺序为 min_lon min_lat max_lon max_lat
--no-clean	不做基础清洗，保留原始错误值
--strict	多文件导入时任一文件失败即报错
--out-npz	输出统一降水中间文件
--json	JSON 输出摘要

XML 工作流

pyufim run-xml config\pyufim_workflow.xml --json

Python 调用 CLI

接口	用途
pyufim.cli.main.build_parser() -> argparse.ArgumentParser	创建命令行参数解析器
pyufim.cli.main.main(argv=None) -> int	执行命令行入口，返回进程退出码

原生库编译

当前原生源码在 native_src/，编译产物放在 bin/：

目录	输出	用途
native_src/ufim-catchment	ufimcatchment.dll / ufimcatchment.so	D8/DINF
native_src/ufim-data	ufimdata.dll / libufimdata.so	数据清洗和校正

Windows MinGW 示例：

cmake -S native_src\ufim-catchment -B native_src\ufim-catchment\build -G "MinGW Makefiles"
cmake --build native_src\ufim-catchment\build --config Release
cmake -S native_src\ufim-data -B native_src\ufim-data\build -G "MinGW Makefiles"
cmake --build native_src\ufim-data\build --config Release

Linux 示例：

cmake -S native_src/ufim-catchment -B native_src/ufim-catchment/build
cmake --build native_src/ufim-catchment/build --config Release
cmake -S native_src/ufim-data -B native_src/ufim-data/build
cmake --build native_src/ufim-data/build --config Release

当前实现状态和限制

功能	状态
D8/DINF 流向、累积、阈值分区	已实现，原生优先
ASC DEM 输入输出	已实现
NetCDF DEM 输入输出	已实现，需要 xarray/netCDF4
降水多格式导入	已实现 CSV/TXT/TSV、NetCDF、GRIB、MICAPS 000/DAT；NetCDF/GRIB 依赖可选包
降水错误值处理	已实现基础清洗、规则校正、自定义回调和“不清洗直接导入”；基础清洗默认优先走 ufimdata C native
通用数据准备	已实现 prepare_dataset，当前支持降水和 ASC 栅格
时间空间裁剪	降水支持时间和空间；ASC 栅格支持空间裁剪
数据 native 后端	已实现 ufimdata.dll，同一 CMake 工程可在 Linux 生成 SO
可视化	已实现曲线、栅格 PNG、GIF/MP4/MOV/AVI 生成入口，依赖 pyufim[plot]
10 亿格级别目标	原生分区算法已按线性复杂度实现，但还缺少端到端大文件 IO、分块和原生 DEM 预处理验证
DEM 填洼/去平	API 已保留，原生实现未完成
SCS 二维面产流	已实现
LISFLOOD 二维	原生接口已封装，依赖 DLL/SO 能力
RFSM	接口已暴露，求解器未实现
紧耦合	接口已暴露，完整求解流程未实现
参数率定	已实现 grid/random/scipy 差分进化率定、轻量敏感性分析和不确定性集合

官方资料来源

资料	链接
UFIM 官网	https://www.antmap.net/web/ufim/
UFIM Desktop 2.0 操作手册	https://oss.antmap.net/software/ufim/resource/2.0/UFIM%20Desktop2.0%E8%BD%AF%E4%BB%B6%E6%93%8D%E4%BD%9C%E6%89%8B%E5%86%8C.pdf
样例数据 sample1	https://oss.antmap.net/software/ufim/resource/2.0/%E6%A0%B7%E4%BE%8B%E6%95%B0%E6%8D%AEsample1.zip
样例数据 sample2	https://oss.antmap.net/software/ufim/resource/2.0/%E6%A0%B7%E4%BE%8B%E6%95%B0%E6%8D%AEsample2.zip
样例数据 sample3	https://oss.antmap.net/software/ufim/resource/2.0/%E6%A0%B7%E4%BE%8B%E6%95%B0%E6%8D%AEsample3.zip