Skip to content

models.hydro

Hydrological model registry.

This module provides access to hydrological models implemented in the holmes_rs Rust extension.

get_config 

get_config(model: HydroModel) -> list[dict[str, str | float]]

Get model parameter configuration.

Parameters:

Name Type Description Default
model HydroModel

Model name (see HydroModel for valid options)

 required

Returns:

Type Description
list[dict]

List of parameter configurations with name, default, min, max, description
Source code in src/holmes/models/hydro.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_config(model: HydroModel) -> list[dict[str, str | float]]:
    """
    Get model parameter configuration.

    Parameters
    ----------
    model : HydroModel
        Model name (see HydroModel for valid options)

    Returns
    -------
    list[dict]
        List of parameter configurations with name, default, min, max,
        description
    """
    try:
        match model:
            case "gr4j":
                param_names = gr4j.param_names
                descriptions = gr4j.param_descriptions
                defaults, bounds = gr4j.init()
            case "bucket":
                param_names = bucket.param_names
                descriptions = bucket.param_descriptions
                defaults, bounds = bucket.init()
            case "cequeau":
                param_names = cequeau.param_names
                descriptions = cequeau.param_descriptions
                defaults, bounds = cequeau.init()
            case _:  # pragma: no cover
                assert_never(model)  # type: ignore
    except (HolmesNumericalError, HolmesValidationError) as exc:
        logger.error(f"Failed to initialize {model} model: {exc}")
        raise
    except Exception as exc:  # pragma: no cover
        logger.exception(f"Unexpected error initializing {model} model")
        raise HolmesError(f"Failed to initialize model: {exc}") from exc

    return [
        {
            "name": name,
            "default": default,
            "min": bounds_[0],
            "max": bounds_[1],
            "description": desc,
        }
        for name, desc, default, bounds_ in zip(
            param_names, descriptions, defaults, bounds
        )
    ]

get_model 

get_model(
    model: HydroModel,
) -> Callable[
    [NDArray[float64], NDArray[float64], NDArray[float64]], NDArray[float64]
]

Get a wrapped model simulation function.

The returned function wraps the underlying Rust implementation with error handling and logging.

Parameters:

Name Type Description Default
model HydroModel

Model name (see HydroModel for valid options)

 required

Returns:

Type Description
Callable

Simulation function that takes (params, precipitation, pet) and returns streamflow
Source code in src/holmes/models/hydro.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_model(
    model: HydroModel,
) -> Callable[
    [
        npt.NDArray[np.float64],
        npt.NDArray[np.float64],
        npt.NDArray[np.float64],
    ],
    npt.NDArray[np.float64],
]:
    """
    Get a wrapped model simulation function.

    The returned function wraps the underlying Rust implementation
    with error handling and logging.

    Parameters
    ----------
    model : HydroModel
        Model name (see HydroModel for valid options)

    Returns
    -------
    Callable
        Simulation function that takes (params, precipitation, pet)
        and returns streamflow
    """
    match model:
        case "gr4j":
            simulate_fn = gr4j.simulate
        case "bucket":
            simulate_fn = bucket.simulate
        case "cequeau":
            simulate_fn = cequeau.simulate
        case _:  # pragma: no cover
            assert_never(model)  # type: ignore

    def wrapped_simulate(
        params: npt.NDArray[np.float64],
        precipitation: npt.NDArray[np.float64],
        pet: npt.NDArray[np.float64],
    ) -> npt.NDArray[np.float64]:
        """Wrapped simulation function with error handling."""
        try:
            return simulate_fn(params, precipitation, pet)
        except (HolmesNumericalError, HolmesValidationError) as exc:
            logger.error(f"Simulation failed for {model}: {exc}")
            raise
        except Exception as exc:  # pragma: no cover
            logger.exception(f"Unexpected error in {model} simulation")
            raise HolmesError(f"Simulation failed: {exc}") from exc

    return wrapped_simulate