precipitation

Gridded precipitation GeoTIFF export from HEC-RAS plan and unsteady HDF files.

API Reference

ras2cng.precipitation.PrecipitationGridInfo dataclass

Metadata for a gridded precipitation dataset inside a HEC-RAS HDF file.

Source code in ras2cng/precipitation.py

@dataclass(frozen=True)
class PrecipitationGridInfo:
    """Metadata for a gridded precipitation dataset inside a HEC-RAS HDF file."""

    hdf_path: Path
    source: str
    values_path: str
    timestamps: list[str]
    rows: int
    cols: int
    cellsize: float
    left: float
    top: float
    projection: str | None = None
    units: str | None = None
    data_type: str | None = None
    nodata: float | None = None
    source_is_cumulative: bool = False

ras2cng.precipitation.PrecipitationExportResult dataclass

Files written by :func:export_precipitation_rasters.

Source code in ras2cng/precipitation.py

@dataclass
class PrecipitationExportResult:
    """Files written by :func:`export_precipitation_rasters`."""

    source_hdf: Path
    source: str
    values_path: str
    output_dir: Path
    units: str | None
    rows: int
    cols: int
    timestamps: list[str] = field(default_factory=list)
    incremental: list[Path] = field(default_factory=list)
    cumulative: list[Path] = field(default_factory=list)

ras2cng.precipitation.list_precipitation_timestamps(hdf_path, *, source='auto')

List gridded precipitation timestamps in a HEC-RAS HDF file.

Parameters:

Name	Type	Description	Default
hdf_path	Path	Plan HDF (*.p??.hdf) or unsteady-flow HDF (*.u??.hdf) containing gridded precipitation.	required
source	PrecipitationSource	Which precipitation dataset to inspect. "processed" uses Precipitation/Values from completed plan HDF files. "imported" uses Imported Raster Data/Values. "auto" prefers processed values and falls back to imported values.	'auto'

Returns:

Type	Description
list[str]	Timestamp labels as stored in the HDF.

Source code in ras2cng/precipitation.py

def list_precipitation_timestamps(
    hdf_path: Path,
    *,
    source: PrecipitationSource = "auto",
) -> list[str]:
    """List gridded precipitation timestamps in a HEC-RAS HDF file.

    Args:
        hdf_path: Plan HDF (``*.p??.hdf``) or unsteady-flow HDF
            (``*.u??.hdf``) containing gridded precipitation.
        source: Which precipitation dataset to inspect. ``"processed"`` uses
            ``Precipitation/Values`` from completed plan HDF files. ``"imported"``
            uses ``Imported Raster Data/Values``. ``"auto"`` prefers processed
            values and falls back to imported values.

    Returns:
        Timestamp labels as stored in the HDF.
    """

    return read_precipitation_grid_info(hdf_path, source=source).timestamps

ras2cng.precipitation.read_precipitation_grid_info(hdf_path, *, source='auto')

Read gridded precipitation raster metadata from a HEC-RAS HDF file.

The HDF must include raster georeferencing attributes: Raster Rows, Raster Cols, Raster Cellsize, Raster Left, and Raster Top. Projection WKT and units are preserved when present.

Source code in ras2cng/precipitation.py

def read_precipitation_grid_info(
    hdf_path: Path,
    *,
    source: PrecipitationSource = "auto",
) -> PrecipitationGridInfo:
    """Read gridded precipitation raster metadata from a HEC-RAS HDF file.

    The HDF must include raster georeferencing attributes: ``Raster Rows``,
    ``Raster Cols``, ``Raster Cellsize``, ``Raster Left``, and ``Raster Top``.
    Projection WKT and units are preserved when present.
    """

    hdf_path = Path(hdf_path)
    with h5py.File(hdf_path, "r") as hdf:
        if PRECIPITATION_GROUP not in hdf:
            raise ValueError(f"No gridded precipitation group found: {PRECIPITATION_GROUP}")

        values_path, resolved_source = _resolve_values_path(hdf, source)
        group = hdf[PRECIPITATION_GROUP]
        dataset = hdf[values_path]
        attrs = _merged_attrs(group.attrs, dataset.attrs)

        rows = _required_int_attr(attrs, "Raster Rows")
        cols = _required_int_attr(attrs, "Raster Cols")
        expected = rows * cols
        if len(dataset.shape) < 2 or int(dataset.shape[-1]) != expected:
            raise ValueError(
                f"Precipitation values at {values_path!r} have shape {dataset.shape}; "
                f"expected the final dimension to equal Raster Rows x Raster Cols ({expected})."
            )

        timestamps = _read_timestamps(hdf, attrs, int(dataset.shape[0]))

        return PrecipitationGridInfo(
            hdf_path=hdf_path,
            source=resolved_source,
            values_path=values_path,
            timestamps=timestamps,
            rows=rows,
            cols=cols,
            cellsize=_required_float_attr(attrs, "Raster Cellsize"),
            left=_required_float_attr(attrs, "Raster Left"),
            top=_required_float_attr(attrs, "Raster Top"),
            projection=_optional_str_attr(attrs, "Projection"),
            units=_optional_str_attr(attrs, "Units"),
            data_type=_optional_str_attr(attrs, "Data Type"),
            nodata=_optional_float_attr(attrs, "NoData"),
            source_is_cumulative=_is_cumulative_source(attrs),
        )

ras2cng.precipitation.export_precipitation_rasters(hdf_path, output_dir, *, source='auto', timestamps=None, units='native', export_incremental=True, export_cumulative=True, prefix=None, overwrite=True, compress='deflate')

Export gridded precipitation GeoTIFF rasters from a HEC-RAS HDF file.

Processed plan-HDF precipitation values are stored as timestep amounts in the HDF precipitation units. Imported raster data may be cumulative source data; in that case incremental rasters are derived by differencing adjacent cumulative grids.

Parameters:

Name	Type	Description	Default
hdf_path	Path	Plan HDF (*.p??.hdf) or unsteady-flow HDF.	required
output_dir	Path	Directory where GeoTIFFs will be written.	required
source	PrecipitationSource	"auto" prefers processed plan results and falls back to imported raster data. Use "processed" or "imported" to require a specific dataset.	'auto'
timestamps	Sequence[str | int] | None	Optional timestamp labels or integer indices to export. Each token is matched against the actual timestamp labels first; only a token with no matching label is interpreted as a zero-based integer index. None exports every timestep.	None
units	PrecipitationUnits	Output units. "native" (default) preserves the HDF source units unchanged. "in" or "mm" converts raster values to the requested unit (mm to in divides by 25.4, in to mm multiplies by 25.4) and tags the GeoTIFF with the target unit. Conversion is skipped if the source units already match the requested unit.	'native'
export_incremental	bool	Write per-timestep precipitation amount rasters.	True
export_cumulative	bool	Write cumulative-through-timestep rasters.	True
prefix	str | None	Optional filename prefix. Defaults to the HDF stem.	None
overwrite	bool	If False, raise when an output file already exists.	True
compress	str	GeoTIFF compression setting passed to rasterio.	'deflate'

Returns:

Name	Type	Description
A	PrecipitationExportResult	class:PrecipitationExportResult describing the written files.

Source code in ras2cng/precipitation.py

def export_precipitation_rasters(
    hdf_path: Path,
    output_dir: Path,
    *,
    source: PrecipitationSource = "auto",
    timestamps: Sequence[str | int] | None = None,
    units: PrecipitationUnits = "native",
    export_incremental: bool = True,
    export_cumulative: bool = True,
    prefix: str | None = None,
    overwrite: bool = True,
    compress: str = "deflate",
) -> PrecipitationExportResult:
    """Export gridded precipitation GeoTIFF rasters from a HEC-RAS HDF file.

    Processed plan-HDF precipitation values are stored as timestep amounts in
    the HDF precipitation units. Imported raster data may be cumulative source
    data; in that case incremental rasters are derived by differencing adjacent
    cumulative grids.

    Args:
        hdf_path: Plan HDF (``*.p??.hdf``) or unsteady-flow HDF.
        output_dir: Directory where GeoTIFFs will be written.
        source: ``"auto"`` prefers processed plan results and falls back to
            imported raster data. Use ``"processed"`` or ``"imported"`` to
            require a specific dataset.
        timestamps: Optional timestamp labels or integer indices to export.
            Each token is matched against the actual timestamp labels first;
            only a token with no matching label is interpreted as a zero-based
            integer index. ``None`` exports every timestep.
        units: Output units. ``"native"`` (default) preserves the HDF source
            units unchanged. ``"in"`` or ``"mm"`` converts raster values to the
            requested unit (mm to in divides by 25.4, in to mm multiplies by
            25.4) and tags the GeoTIFF with the target unit. Conversion is
            skipped if the source units already match the requested unit.
        export_incremental: Write per-timestep precipitation amount rasters.
        export_cumulative: Write cumulative-through-timestep rasters.
        prefix: Optional filename prefix. Defaults to the HDF stem.
        overwrite: If False, raise when an output file already exists.
        compress: GeoTIFF compression setting passed to rasterio.

    Returns:
        A :class:`PrecipitationExportResult` describing the written files.
    """

    if not export_incremental and not export_cumulative:
        raise ValueError("At least one of export_incremental or export_cumulative must be True")

    try:
        import rasterio
        from rasterio.crs import CRS
        from rasterio.transform import from_origin
    except ImportError as exc:  # pragma: no cover - exercised only without rasterio installed
        raise ImportError(
            "export_precipitation_rasters() requires rasterio. "
            'Install ras2cng with the "all" or "pmtiles" extra, or install rasterio separately.'
        ) from exc

    hdf_path = Path(hdf_path)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    info = read_precipitation_grid_info(hdf_path, source=source)
    selected_indices = _select_indices(info.timestamps, timestamps)
    file_prefix = _safe_filename_part(prefix or hdf_path.stem)

    with h5py.File(hdf_path, "r") as hdf:
        raw = np.asarray(hdf[info.values_path][:], dtype="float32")

    grids = _reshape_values(raw, info)
    incremental, cumulative = _incremental_and_cumulative(grids, info)

    factor, target_units = _unit_conversion(info.units, units)
    if factor != 1.0:
        # NaN cells (nodata) are preserved through multiplication; the nodata
        # sentinel in ``info.nodata`` is left untouched so it round-trips.
        incremental = (incremental.astype("float64") * factor).astype("float32")
        cumulative = (cumulative.astype("float64") * factor).astype("float32")
    if target_units is not None:
        info = replace(info, units=target_units)

    transform = from_origin(info.left, info.top, info.cellsize, info.cellsize)
    crs = _rasterio_crs(CRS, info.projection)

    profile = {
        "driver": "GTiff",
        "height": info.rows,
        "width": info.cols,
        "count": 1,
        "dtype": "float32",
        "transform": transform,
        "compress": compress,
    }
    if crs is not None:
        profile["crs"] = crs
    if info.nodata is not None:
        profile["nodata"] = info.nodata

    result = PrecipitationExportResult(
        source_hdf=hdf_path,
        source=info.source,
        values_path=info.values_path,
        output_dir=output_dir,
        units=info.units,
        rows=info.rows,
        cols=info.cols,
    )

    for index in selected_indices:
        timestamp = info.timestamps[index]
        stamp = _safe_timestamp_part(timestamp)
        result.timestamps.append(timestamp)

        if export_incremental:
            out = output_dir / f"{file_prefix}_precip_{index:04d}_{stamp}.tif"
            _write_geotiff(
                rasterio,
                out,
                incremental[index],
                profile,
                info,
                timestamp,
                "incremental",
                overwrite=overwrite,
            )
            result.incremental.append(out)

        if export_cumulative:
            out = output_dir / f"{file_prefix}_precip_cumulative_{index:04d}_{stamp}.tif"
            _write_geotiff(
                rasterio,
                out,
                cumulative[index],
                profile,
                info,
                timestamp,
                "cumulative",
                overwrite=overwrite,
            )
            result.cumulative.append(out)

    return result