derivations.pywatershed

pywatershed

pywatershed parameter derivation plugin.

Convert SIR physical properties (zonal statistics of raw geospatial data) into PRMS/pywatershed model parameters. This module implements the model-specific derivation pipeline defined in docs/reference/pywatershed_dataset_param_map.yml, transforming source-unit geospatial statistics into the internal unit system required by PRMS (feet, inches, degrees Fahrenheit, acres).

The derivation follows a 15-step DAG. Steps implemented here:

1. Geometry --- HRU area (acres) and latitude (decimal degrees)
2. Topology --- segment routing (tosegment, hru_segment, seg_length)
3. Topography --- elevation (feet), slope (decimal fraction), aspect (degrees) 3b. Segment elevation --- mean channel elevation from 3DEP DEM via InterpGen
4. Land cover --- NLCD reclassification to PRMS cov_type, canopy density, imperviousness
5. Soils --- gNATSGO/STATSGO2 texture classification and AWC
6. Waterbody --- NHDPlus depression storage overlay
7. Forcing --- temporal CBH merge (prcp, tmax, tmin)
8. Lookup tables --- interception capacities and winter cover density
9. Soltab --- potential solar radiation tables (Swift 1976)
10. PET --- Jensen-Haise coefficients from climate normals
11. Transpiration --- frost-free period timing from monthly tmin
12. Routing --- Muskingum routing parameters (K_coef, x_coef, seg_slope, etc.)
13. Defaults --- standard PRMS default values and initial conditions
14. Calibration seeds --- physically-based initial values for calibration parameters

References

Markstrom, S. L., et al. (2015). PRMS-IV, the Precipitation-Runoff Modeling System, Version 4. USGS Techniques and Methods 6-B7. Regan, R. S., et al. (2018). Description of the National Hydrologic Model Infrastructure. USGS Techniques and Methods 6-B9.

See Also

hydro_param.plugins.DerivationContext : Typed input bundle for derivation. hydro_param.units.convert : Unit conversion dispatch used throughout. hydro_param.solar.compute_soltab : Solar radiation table computation.

PywatershedDerivation

PywatershedDerivation()

Derive pywatershed/PRMS parameters from SIR physical properties.

Implement the full derivation pipeline defined in docs/reference/pywatershed_dataset_param_map.yml, converting source-unit zonal statistics from the Standardized Internal Representation (SIR) into the ~100+ parameters required by pywatershed (PRMS-IV in Python).

The derivation follows a directed acyclic graph (DAG) of 15 ordered steps. Each step is implemented as a private method (e.g., _derive_geometry for step 1). Steps execute in dependency order: later steps may read parameters produced by earlier ones (e.g., step 8 reads cov_type from step 4).

This class conforms to the DerivationPlugin protocol defined in hydro_param.plugins.

ATTRIBUTE	DESCRIPTION
name	Plugin identifier ("pywatershed"). TYPE: str

Notes

PRMS internal units are: feet, inches, degrees Fahrenheit, acres. All unit conversions from SI source data happen within individual derivation steps using hydro_param.units.convert.

Lookup tables are loaded lazily from YAML files and cached in _lookup_cache for the lifetime of the instance.

References

Markstrom, S. L., et al. (2015). PRMS-IV, the Precipitation-Runoff Modeling System, Version 4. USGS Techniques and Methods 6-B7.

See Also

hydro_param.plugins.DerivationContext : Input bundle for derive(). hydro_param.plugins.DerivationPlugin : Protocol this class implements. hydro_param.formatters.pywatershed : Output formatter for pywatershed.

Source code in src/hydro_param/derivations/pywatershed.py

def __init__(self) -> None:
    self._lookup_cache: dict[str, dict] = {}

derive

derive(context: DerivationContext) -> xr.Dataset

Derive all pywatershed parameters from the SIR.

Execute the full derivation DAG in dependency order, producing a single xr.Dataset containing all derivable PRMS parameters. Steps that lack required input data log warnings and are skipped gracefully. Parameter overrides from the config are applied last.

PARAMETER	DESCRIPTION
context	Typed input bundle containing the SIR dataset, target fabric GeoDataFrame, segment GeoDataFrame, waterbody GeoDataFrame, temporal forcing datasets, lookup table directory, and pipeline configuration. temporal may be None if no temporal SIR data is available, in which case step 7 (forcing) is skipped and PET/transpiration steps use scalar defaults. TYPE: DerivationContext

RETURNS	DESCRIPTION
Dataset	Parameter dataset with PRMS-convention variable names and units. Dimensions are nhru (and nsegment for routing, ndoy for soltab, nmonth for monthly parameters, time for forcing).

Notes

Step execution order: 1 (geometry) -> 2 (topology) -> 3 (topo) -> 3b (seg_elev) -> 4 (landcover) -> 5 (soils) -> 6 (waterbody) -> 8 (lookups) -> 12 (routing) -> 9 (soltab) -> 10 (PET) -> 11 (transp) -> 13 (defaults) -> 14 (calibration) -> 7 (forcing) -> overrides.

Step 7 (forcing) runs late because it has no downstream dependencies within the static parameter DAG.

Source code in src/hydro_param/derivations/pywatershed.py

def derive(self, context: DerivationContext) -> xr.Dataset:
    """Derive all pywatershed parameters from the SIR.

    Execute the full derivation DAG in dependency order, producing a
    single ``xr.Dataset`` containing all derivable PRMS parameters.
    Steps that lack required input data log warnings and are skipped
    gracefully.  Parameter overrides from the config are applied last.

    Parameters
    ----------
    context : DerivationContext
        Typed input bundle containing the SIR dataset, target fabric
        GeoDataFrame, segment GeoDataFrame, waterbody GeoDataFrame,
        temporal forcing datasets, lookup table directory, and
        pipeline configuration.  ``temporal`` may be ``None`` if no
        temporal SIR data is available, in which case step 7 (forcing)
        is skipped and PET/transpiration steps use scalar defaults.

    Returns
    -------
    xr.Dataset
        Parameter dataset with PRMS-convention variable names and units.
        Dimensions are ``nhru`` (and ``nsegment`` for routing, ``ndoy``
        for soltab, ``nmonth`` for monthly parameters, ``time`` for
        forcing).

    Notes
    -----
    Step execution order: 1 (geometry) -> 2 (topology) -> 3 (topo) ->
    3b (seg_elev) -> 4 (landcover) -> 5 (soils) -> 6 (waterbody) ->
    8 (lookups) -> 12 (routing) -> 9 (soltab) -> 10 (PET) ->
    11 (transp) -> 13 (defaults) -> 14 (calibration) -> 7 (forcing) ->
    overrides.

    Step 7 (forcing) runs late because it has no downstream
    dependencies within the static parameter DAG.
    """
    sir = context.sir
    id_field = context.fabric_id_field
    fabric = context.fabric

    # Derive nhru count and IDs from fabric (authoritative source).
    # Fall back to SIR first variable length when no fabric is provided
    # (e.g., library API use without a fabric file).
    if fabric is not None and id_field in fabric.columns:
        nhru = len(fabric)
        hru_ids = fabric[id_field].values
    elif sir.data_vars:
        first_var = sir.data_vars[0]
        first_da = sir[first_var]
        nhru = len(first_da)
        # Try to recover HRU IDs from the SIR variable's coordinates.
        if id_field in first_da.dims and id_field in first_da.coords:
            hru_ids = first_da.coords[id_field].values
        else:
            hru_ids = None
    else:
        raise ValueError(
            f"Cannot determine HRU count: fabric is None or missing "
            f"'{id_field}' column, and SIR contains no static variables. "
            f"Provide a fabric GeoDataFrame with an '{id_field}' column."
        )

    ds = xr.Dataset()

    # Carry HRU coordinates so derived params retain stable indexing
    if hru_ids is not None:
        ds = ds.assign_coords(nhru=hru_ids)

    # Step 1: Geometry (hru_area, hru_lat)
    ds = self._derive_geometry(context, ds)

    # Step 2: Topology (tosegment, hru_segment, seg_length)
    ds = self._derive_topology(context, ds)

    # Step 3: Topographic parameters (hru_elev, hru_slope, hru_aspect)
    ds = self._derive_topography(context, ds)

    # Step 3b: Segment elevation (InterpGen + 3DEP DEM)
    ds = self._derive_segment_elevation(context, ds)

    # Step 4: Land cover parameters (cov_type, covden_sum, hru_percent_imperv)
    ds = self._derive_landcover(context, ds)

    # Step 5: Soils parameters (soil_type, soil_moist_max, soil_rechr_max_frac)
    ds = self._derive_soils(context, ds)

    # Step 6: Waterbody overlay (dprst_frac, hru_type)
    ds = self._derive_waterbody(context, ds)

    # Step 8: Lookup table application
    ds = self._apply_lookup_tables(context, ds)

    # Step 12: Routing parameters
    ds = self._derive_routing(context, ds)

    # Step 9: Solar radiation tables (soltab)
    ds = self._derive_soltab(context, ds)

    # Compute monthly climate normals once for steps 10 and 11
    normals = self._compute_monthly_normals(context)

    # Step 10: PET coefficients (Jensen-Haise)
    ds = self._derive_pet_coefficients(ds, normals)

    # Step 11: Transpiration timing (frost-free period)
    ds = self._derive_transp_timing(ds, normals)

    # Step 13: Defaults and initial conditions
    ds = self._apply_defaults(ds, nhru, context)

    # Step 14: Calibration seeds
    ds = self._derive_calibration_seeds(context, ds)

    # Step 7: Forcing (temporal merge — runs late, no downstream deps)
    ds = self._derive_forcing(context, ds)

    # Apply parameter overrides last
    overrides = context.config.get("parameter_overrides", {})
    if isinstance(overrides, dict) and "values" in overrides:
        overrides = overrides["values"]
    if overrides:
        ds = self._apply_overrides(ds, overrides)

    return ds