"""Adapter that wraps the existing ``UnitHydrograph`` class as an ``IModel``.
The ``UnitHydrograph`` class (from ``uh_models.py``) has its own internal
registry, parameter dict, convolution, and calibration logic. Rather than
reimplementing any of that, this module bridges it to the sparsehydro
lifecycle and parameter registry via the **Adapter pattern**.
Usage::
from sparsehydro.models.unithydrograph import register_all_uh_models
# Call once after uh_models is on sys.path
register_all_uh_models()
from sparsehydro.registry import registry
model = registry.create("uh-nash")
model.initialize()
model.validate()
model.prepare(rain_stormflow_df)
result = model.predict()
model.finalize()
``UnitHydrograph`` is imported **lazily** — ``import sparsehydro`` always
succeeds; the error surfaces only when you first call a method that needs the
wrapped class. Ensure ``uh_models.py`` and its dependencies
(``data_processing``, ``objective_functions``) are on ``sys.path`` before
calling :func:`register_all_uh_models` or :func:`create_uh_model`.
"""
from __future__ import annotations
from typing import Any, ClassVar
import numpy as np
import pandas as pd
from ...enums import ModelState
from ..base import IUnitHydroComponent
from ...parameters import ScalarParameter
from ...registry import registry
# ---------------------------------------------------------------------------
# Lazy import helpers
# ---------------------------------------------------------------------------
_UnitHydrograph: type | None = None
_INF_SUBSTITUTE = 1e9 # finite stand-in for np.inf in ScalarParameter bounds
def _get_uh_class() -> type:
"""Return the ``UnitHydrograph`` class, importing it on first call.
:returns: The lazily-imported ``UnitHydrograph`` class.
:rtype: type
:raises ImportError: If ``uh_models`` is not importable.
"""
global _UnitHydrograph
if _UnitHydrograph is None:
try:
from uh_models import UnitHydrograph # type: ignore[import]
_UnitHydrograph = UnitHydrograph
except ImportError as exc:
raise ImportError(
"uh_models is required for UnitHydrographAdapter. "
"Ensure uh_models.py and its dependencies "
"(data_processing, objective_functions) are on sys.path."
) from exc
return _UnitHydrograph
# ---------------------------------------------------------------------------
# Adapter base class
# ---------------------------------------------------------------------------
[docs]
class UnitHydrographAdapter(IUnitHydroComponent):
"""Adapter base that bridges ``UnitHydrograph`` to the ``IModel`` lifecycle.
**Do not instantiate or register this class directly.**
Use :func:`create_uh_model` or :func:`register_all_uh_models` to obtain
concrete, registry-compatible subclasses.
**Lifecycle mapping**
+-----------------+--------------------------------------------------+
| IModel method | Adapter action |
+=================+==================================================+
| ``initialize`` | Constructs ``UnitHydrograph(key)``, reads |
| | ``_registry`` to create one ``ScalarParameter`` |
| | per model parameter. |
+-----------------+--------------------------------------------------+
| ``validate`` | Delegates to ``parameters_valid()``. |
+-----------------+--------------------------------------------------+
| ``prepare`` | Stores the ``rain_stormflow`` DataFrame; syncs |
| | ``ScalarParameter`` values → ``_uh.parameters``. |
+-----------------+--------------------------------------------------+
| ``predict`` | Calls ``_uh.predict()``; returns the DataFrame. |
+-----------------+--------------------------------------------------+
| ``finalize`` | Releases the stored DataFrame. |
+-----------------+--------------------------------------------------+
**Parameter synchronisation**
``ScalarParameter`` values are the source of truth. They are pushed to
``_uh.parameters`` before every ``predict`` or ``fit`` call. After
``fit`` the optimised values are pulled back so both representations stay
consistent.
**Infinite bounds**
``UnitHydrograph._registry`` uses ``np.inf`` for unbounded parameters
(e.g. ``A``). These are clamped to ±:data:`_INF_SUBSTITUTE` when
creating ``ScalarParameter`` objects so that calibration algorithms that
require finite bounds always receive them.
"""
model_name: ClassVar[str] = "uh-adapter"
_uh_model_name: ClassVar[str] = ""
_amplitude_param_name: ClassVar[str | None] = "A"
def __init__(self) -> None:
super().__init__()
self._uh: Any = None
self._prepared_data: pd.DataFrame | None = None
# ------------------------------------------------------------------
# IModel lifecycle
# ------------------------------------------------------------------
[docs]
def initialize(self) -> None:
"""Construct the wrapped ``UnitHydrograph`` and register its parameters.
:returns: Nothing.
:rtype: None
:raises TypeError: If called on the base ``UnitHydrographAdapter``
instead of a factory-created subclass.
"""
UH = _get_uh_class()
uh_key = type(self)._uh_model_name
if not uh_key:
raise TypeError(
f"{type(self).__name__} has no _uh_model_name set. "
"Use create_uh_model() or register_all_uh_models() to obtain "
"a concrete adapter class."
)
self._uh = UH(uh_key)
model_def = UH._registry[uh_key]
for name, p0, lb, ub in zip(
model_def["pnames"],
model_def["p0"],
model_def["lb"],
model_def["ub"],
):
lb_f = float(lb) if not np.isinf(lb) else -_INF_SUBSTITUTE
ub_f = float(ub) if not np.isinf(ub) else _INF_SUBSTITUTE
self.register_scalar_parameter(
ScalarParameter(
name=name,
value=float(p0),
lower_bound=lb_f,
upper_bound=ub_f,
)
)
self._state = ModelState.INITIALIZED
[docs]
def validate(self) -> bool:
"""Validate that all registered parameters satisfy their bounds.
:returns: ``True`` if all parameters are within bounds.
:rtype: bool
"""
ok = self.parameters_valid()
if ok:
self._state = ModelState.VALIDATED
return ok
[docs]
def prepare(
self,
rain_stormflow: pd.DataFrame,
**kwargs: Any,
) -> None:
"""Store the input DataFrame and sync parameters to the wrapped model.
:param rain_stormflow: Rainfall/stormflow forcing DataFrame consumed by
the wrapped ``UnitHydrograph``.
:type rain_stormflow: pandas.DataFrame
:param kwargs: Reserved for API compatibility; ignored.
:type kwargs: Any
:returns: Nothing.
:rtype: None
"""
self._prepared_data = rain_stormflow
self._sync_to_uh()
self._state = ModelState.PREPARED
[docs]
def predict(
self,
predict_range: Any = None,
trim: bool = True,
) -> pd.DataFrame:
"""Convolve stored rainfall with the UH shape and return predicted flow.
:param predict_range: Optional prediction window forwarded to the
wrapped model's ``predict``.
:type predict_range: Any
:param trim: Whether to trim the convolution output to the input length.
:type trim: bool
:returns: Predicted-flow DataFrame from the wrapped model.
:rtype: pandas.DataFrame
:raises RuntimeError: If :meth:`initialize` or :meth:`prepare` has not
been called.
"""
if self._uh is None or self._prepared_data is None:
raise RuntimeError(
"Call initialize() and prepare(rain_stormflow) before predict()."
)
self._sync_to_uh()
result: pd.DataFrame = self._uh.predict(
self._prepared_data, predict_range=predict_range, trim=trim
)
self._state = ModelState.PREDICTED
return result
[docs]
def finalize(self) -> None:
"""Release the stored DataFrame and advance state to ``FINALIZED``.
:returns: Nothing.
:rtype: None
"""
self._prepared_data = None
self._state = ModelState.FINALIZED
# ------------------------------------------------------------------
# Calibration passthrough
# ------------------------------------------------------------------
[docs]
def fit(
self,
rain_stormflow: pd.DataFrame,
**kwargs: Any,
) -> dict[str, float]:
"""Calibrate the wrapped ``UnitHydrograph`` and sync results back.
:param rain_stormflow: Rainfall/stormflow forcing DataFrame.
:type rain_stormflow: pandas.DataFrame
:param kwargs: Additional keyword arguments forwarded to the wrapped
model's ``fit``.
:type kwargs: Any
:returns: Mapping of fitted parameter name to value.
:rtype: dict[str, float]
:raises RuntimeError: If :meth:`initialize` has not been called.
"""
if self._uh is None:
raise RuntimeError("Call initialize() before fit().")
self._sync_to_uh()
fitted: dict[str, float] = self._uh.fit(rain_stormflow, **kwargs)
self._sync_from_uh()
return fitted
# ------------------------------------------------------------------
# UH shape access
# ------------------------------------------------------------------
[docs]
def get_uh(self, max_steps: int = 864, norm: int = 0) -> np.ndarray:
"""Return the unit hydrograph ordinate array from the wrapped model.
:param max_steps: Maximum number of ordinates to return.
:type max_steps: int
:param norm: Normalisation mode passed to the wrapped model
(``0`` raw, ``1`` unit-area).
:type norm: int
:returns: UH ordinate array.
:rtype: numpy.ndarray
:raises RuntimeError: If :meth:`initialize` has not been called.
"""
if self._uh is None:
raise RuntimeError("Call initialize() before get_uh().")
self._sync_to_uh()
return self._uh.get_uh(max_steps=max_steps, norm=norm)
[docs]
def get_kernel(
self,
dt_hours: float,
n_steps: int | None = None,
) -> np.ndarray:
"""Return the normalized UH ordinate array for use in a composite model.
Returns ``get_uh(norm=1)`` — the UH shape with the ``A`` amplitude
parameter divided out, so ``np.sum(result) ≈ 1.0``.
.. note::
The ``dt_hours`` argument is accepted for API uniformity but does
not resample the kernel.
:param dt_hours: Time-step size [hr] (accepted for API uniformity).
:type dt_hours: float
:param n_steps: Number of output steps; defaults to the natural support.
:type n_steps: int | None
:returns: Unit-area UH ordinate array.
:rtype: numpy.ndarray
:raises RuntimeError: If :meth:`initialize` has not been called.
"""
if self._uh is None:
raise RuntimeError("Call initialize() before get_kernel().")
self._sync_to_uh()
ordinates = self._uh.get_uh(norm=1)
if n_steps is not None:
if len(ordinates) >= n_steps:
ordinates = ordinates[:n_steps]
else:
ordinates = np.pad(ordinates, (0, n_steps - len(ordinates)))
return ordinates
@property
def is_fit(self) -> bool:
"""``True`` if the wrapped ``UnitHydrograph`` has been successfully fitted.
:returns: Fit status of the wrapped model (``False`` if not yet built).
:rtype: bool
"""
return bool(self._uh.is_fit) if self._uh is not None else False
# ------------------------------------------------------------------
# Parameter synchronisation (private)
# ------------------------------------------------------------------
def _sync_to_uh(self) -> None:
"""Push ``ScalarParameter.value`` → ``_uh.parameters`` dict.
:returns: Nothing.
:rtype: None
"""
if self._uh is None:
return
for name, param in self._scalar_parameters.items():
if name in self._uh.parameters:
self._uh.parameters[name] = param.value
def _sync_from_uh(self) -> None:
"""Pull ``_uh.parameters`` dict → ``ScalarParameter.value``.
:returns: Nothing.
:rtype: None
"""
if self._uh is None:
return
for name, value in self._uh.parameters.items():
if name in self._scalar_parameters:
self._scalar_parameters[name].value = float(value)
# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------
[docs]
def create_uh_model(uh_model_name: str) -> type[UnitHydrographAdapter]:
"""Create and register a concrete ``IModel`` subclass for one UH model.
:param uh_model_name: Key in ``UnitHydrograph._registry``
(e.g. ``"Nash"``, ``"Gamma"``, ``"Weibull"``).
:type uh_model_name: str
:returns: Concrete ``UnitHydrographAdapter`` subclass.
:rtype: type[UnitHydrographAdapter]
:raises KeyError: If ``uh_model_name`` is not in
``UnitHydrograph._registry``.
"""
UH = _get_uh_class()
if uh_model_name not in UH._registry:
raise KeyError(
f"'{uh_model_name}' is not in UnitHydrograph._registry. "
f"Available: {sorted(UH._registry)}"
)
slug = uh_model_name.lower().replace("+", "-").replace("_", "-")
sparse_name = f"uh-{slug}"
cls_name = f"UH_{uh_model_name.replace('+', '_').replace(' ', '_')}"
cls = type(
cls_name,
(UnitHydrographAdapter,),
{
"model_name": sparse_name,
"_uh_model_name": uh_model_name,
"__doc__": (
f"sparsehydro adapter for the ``'{uh_model_name}'`` unit "
f"hydrograph from ``UnitHydrograph._registry``.\n\n"
f"Instantiate via ``registry.create('{sparse_name}')``."
),
},
)
registry.register(cls)
return cls
[docs]
def register_all_uh_models() -> dict[str, type[UnitHydrographAdapter]]:
"""Create and register adapters for **every** model in ``UnitHydrograph._registry``.
Call this once after confirming that ``uh_models`` is importable::
import sys
sys.path.insert(0, "/path/to/UnitHydrograph/Modeling")
from sparsehydro.models.unithydrograph import register_all_uh_models
register_all_uh_models()
Models already registered in the sparsehydro registry are skipped silently.
:returns: Mapping of sparsehydro ``model_name`` → adapter class.
"""
UH = _get_uh_class()
result: dict[str, type[UnitHydrographAdapter]] = {}
for uh_name in UH._registry:
slug = uh_name.lower().replace("+", "-").replace("_", "-")
sparse_name = f"uh-{slug}"
if registry.is_registered(sparse_name):
result[sparse_name] = registry.get(sparse_name) # type: ignore[assignment]
else:
result[sparse_name] = create_uh_model(uh_name)
return result