Source code for pastas.transform

"""This module contains all the transforms that can be added to a model.

These transforms are applied after the simulation, to incorporate nonlinear effects.
"""
import numpy as np
from pandas import DataFrame, Series

# Type Hinting
from pastas.typing import ArrayLike, Model

from .decorators import set_parameter
from .utils import validate_name


[docs]class ThresholdTransform: """ThresholdTransform lowers the simulation when it exceeds a certain value. Parameters ---------- value : float, optional The starting value above which the simulation is lowered. vmin : float, optional The minimum value above which the simulation is lowered. vmin : float, optional The maximum value above which the simulation is lowered. name: str, optional Name of the transform. nparam : int, optional The number of parameters. Default is nparam=2. The first parameter then is the threshold, and the second parameter is the factor with which the simulation is lowered. Notes ----- In geohydrology this transform can be used in a situation where the groundwater level reaches the surface level and forms a lake. Because of the larger storage of the lake, the (groundwater) level then rises slower when it rains. """ _name = "ThresholdTransform"
[docs] def __init__( self, value: float = np.nan, vmin: float = np.nan, vmax: float = np.nan, name: str = "ThresholdTransform", nparam: int = 2, ) -> None: self.value = value self.vmin = vmin self.vmax = vmax self.name = validate_name(name) self.nparam = nparam self.parameters = DataFrame(columns=["initial", "pmin", "pmax", "vary", "name"])
[docs] def set_model(self, ml: Model) -> None: obs = ml.observations() if np.isnan(self.value): self.value = obs.min() + 0.75 * (obs.max() - obs.min()) if np.isnan(self.vmin): self.vmin = obs.min() + 0.5 * (obs.max() - obs.min()) if np.isnan(self.vmax): self.vmax = obs.max() self.set_init_parameters()
[docs] def set_init_parameters(self) -> None: self.parameters.loc[self.name + "_1"] = ( self.value, self.vmin, self.vmax, True, self.name, ) if self.nparam == 2: self.parameters.loc[self.name + "_2"] = (0.5, 0.0, 1.0, True, self.name)
@set_parameter def _set_initial(self, name: str, value: float) -> None: """Internal method to set the initial parameter value. Notes ----- The preferred method for parameter setting is through the model. """ self.parameters.loc[name, "initial"] = value @set_parameter def _set_pmin(self, name: str, value: float) -> None: """Internal method to set the lower bound of the parameter value. Notes ----- The preferred method for parameter setting is through the model. """ self.parameters.loc[name, "pmin"] = value @set_parameter def _set_pmax(self, name: str, value: float) -> None: """Internal method to set the upper bound of the parameter value. Notes ----- The preferred method for parameter setting is through the model. """ self.parameters.loc[name, "pmax"] = value @set_parameter def _set_vary(self, name: str, value: float) -> None: """Internal method to set if the parameter is varied during optimization. Notes ----- The preferred method for parameter setting is through the model. """ self.parameters.loc[name, "vary"] = bool(value)
[docs] def simulate(self, h: Series, p: ArrayLike) -> Series: if self.nparam == 1: # value above a threshold p[0] are equal to the threshold h[h > p[0]] = p[0] elif self.nparam == 2: # values above a threshold p[0] are scaled by p[1] mask = h > p[0] h[mask] = p[0] + p[1] * (h[mask] - p[0]) else: raise ValueError("Not yet implemented yet") return h
[docs] def to_dict(self) -> dict: data = { "class": self._name, "value": self.value, "vmin": self.vmin, "vmax": self.vmax, "name": self.name, "nparam": self.nparam, } return data