Common Utilities and Configurations¶

This section covers common utilities, configurations, and shared components used across all hydrological models in hydromodel.

Model Configuration¶

`ParamRangeConfigError (ValueError)` ¶

Raised when an explicit parameter range configuration is invalid.

Source code in hydromodel/models/model_config.py

class ParamRangeConfigError(ValueError):
    """Raised when an explicit parameter range configuration is invalid."""

`attach_parameter_contract(result, model_setup)` ¶

Add explicit parameter and loss metadata without removing legacy fields.

Source code in hydromodel/models/model_config.py

def attach_parameter_contract(result, model_setup):
    """Add explicit parameter and loss metadata without removing legacy fields."""
    result = dict(result)
    model_name = model_setup.model_name
    best_params = copy.deepcopy(result.get("best_params"))
    normalized = None

    if "objective_value" in result:
        result["objective_value"] = float(result["objective_value"])

    if isinstance(best_params, dict):
        model_params = best_params.get(model_name)
        if isinstance(model_params, dict):
            normalized = {
                name: float(model_params[name])
                for name in model_setup.parameter_names
                if name in model_params
            }
            best_params[model_name] = normalized

    result["parameter_format"] = "normalized"
    result["best_params_normalized"] = (
        {model_name: normalized} if normalized is not None else None
    )
    result["best_params_denormalized"] = (
        {
            model_name: denormalize_parameter_dict(
                normalized, model_setup.model_param_config
            )
        }
        if normalized is not None
        else None
    )
    result["param_range_source"] = model_setup.param_range_source
    result["param_range_source_path"] = model_setup.param_range_source_path
    result["loss_config"] = serializable_loss_config(model_setup.loss_config)
    return result

`denormalize_parameter_dict(parameters, model_param_config)` ¶

Convert a normalized parameter dict to physical values.

Source code in hydromodel/models/model_config.py

def denormalize_parameter_dict(parameters, model_param_config):
    """Convert a normalized parameter dict to physical values."""
    param_names = model_param_config["param_name"]
    param_ranges = model_param_config["param_range"]
    denormalized = OrderedDict()
    for name in param_names:
        value = parameters[name]
        lower, upper = param_ranges[name]
        denormalized[name] = lower + float(value) * (upper - lower)
    return denormalized

`get_model_param_config(model_name, kwargs=None)` ¶

Resolve model parameter metadata from explicit, legacy, or default kwargs.

Source code in hydromodel/models/model_config.py

def get_model_param_config(model_name, kwargs=None):
    """Resolve model parameter metadata from explicit, legacy, or default kwargs."""
    kwargs = kwargs or {}

    param_config = kwargs.get("param_config")
    if isinstance(param_config, dict):
        if model_name in param_config:
            return validate_model_param_dict(
                {model_name: param_config[model_name]}, model_name=model_name
            )[model_name]
        if "param_range" in param_config:
            default_config = MODEL_PARAM_DICT.get(model_name)
            if default_config is None:
                raise ParamRangeConfigError(
                    f"No default parameter configuration for model '{model_name}'"
                )
            names = param_config.get(
                "param_name", default_config["param_name"]
            )
            return validate_model_param_dict(
                {
                    model_name: {
                        "param_name": names,
                        "param_range": param_config["param_range"],
                    }
                },
                model_name=model_name,
            )[model_name]

    explicit_range = kwargs.get("param_range")
    if isinstance(explicit_range, dict):
        default_config = MODEL_PARAM_DICT.get(model_name)
        if default_config is None:
            raise ParamRangeConfigError(
                f"No default parameter configuration for model '{model_name}'"
            )
        names = kwargs.get("param_name", default_config["param_name"])
        return validate_model_param_dict(
            {model_name: {"param_name": names, "param_range": explicit_range}},
            model_name=model_name,
        )[model_name]

    legacy_config = kwargs.get(model_name)
    if isinstance(legacy_config, dict):
        if "param_name" not in legacy_config:
            default_config = MODEL_PARAM_DICT.get(model_name, {})
            legacy_config = dict(
                param_name=default_config.get("param_name", []),
                **legacy_config,
            )
        return validate_model_param_dict(
            {model_name: legacy_config}, model_name=model_name
        )[model_name]

    warnings.warn(
        f"Parameter metadata for model '{model_name}' was not provided. "
        "Falling back to MODEL_PARAM_DICT defaults.",
        RuntimeWarning,
        stacklevel=2,
    )
    return _copy_model_param_dict(MODEL_PARAM_DICT)[model_name]

`inject_model_param_config(model_params, model_name, model_param_config)` ¶

Inject parameter metadata using both explicit and legacy contracts.

Source code in hydromodel/models/model_config.py

def inject_model_param_config(model_params, model_name, model_param_config):
    """Inject parameter metadata using both explicit and legacy contracts."""
    injected = dict(model_params or {})
    injected[model_name] = model_param_config
    injected["param_config"] = {model_name: model_param_config}
    injected["param_name"] = model_param_config["param_name"]
    injected["param_range"] = model_param_config["param_range"]
    return injected

`resolve_model_param_config(model_name, param_range_file=None, fallback_dir=None, strict=False)` ¶

Resolve parameter ranges for a model with explicit source metadata.

Source code in hydromodel/models/model_config.py

def resolve_model_param_config(
    model_name,
    param_range_file=None,
    fallback_dir=None,
    strict=False,
):
    """Resolve parameter ranges for a model with explicit source metadata."""
    source = "default"
    resolved_file = None

    if param_range_file is not None:
        resolved_file = str(param_range_file)
        source = "explicit"
    elif fallback_dir is not None:
        candidate = os.path.join(str(fallback_dir), "param_range.yaml")
        if os.path.exists(candidate):
            resolved_file = candidate
            source = "artifact"

    if resolved_file is None:
        warnings.warn(
            f"param_range_file not provided for model '{model_name}'. "
            "Using default MODEL_PARAM_DICT ranges.",
            RuntimeWarning,
            stacklevel=2,
        )

    param_dict = read_model_param_dict(
        resolved_file, strict=strict or resolved_file is not None
    )
    normalized = validate_model_param_dict(param_dict, model_name=model_name)
    model_param_config = normalized[model_name]

    return {
        "param_dict": param_dict,
        "model_param_config": model_param_config,
        "source": source,
        "source_path": resolved_file,
    }

`serializable_loss_config(loss_config)` ¶

Return a JSON-safe copy of a resolved loss configuration.

Source code in hydromodel/models/model_config.py

def serializable_loss_config(loss_config):
    """Return a JSON-safe copy of a resolved loss configuration."""
    serializable = {}
    for key, value in loss_config.items():
        if callable(value):
            serializable[key] = getattr(value, "__name__", "callable")
        else:
            serializable[key] = value
    return serializable

`validate_model_param_dict(param_dict, model_name=None)` ¶

Validate and normalize model parameter metadata.

The returned param_range order always follows param_name, which is required because normalized parameter vectors are positional.

Source code in hydromodel/models/model_config.py

def validate_model_param_dict(param_dict, model_name=None):
    """Validate and normalize model parameter metadata.

    The returned param_range order always follows param_name, which is required
    because normalized parameter vectors are positional.
    """
    if not isinstance(param_dict, dict):
        raise ParamRangeConfigError(
            "Parameter range file must contain a mapping"
        )

    models = (
        [model_name] if model_name is not None else list(param_dict.keys())
    )
    normalized = {}
    for model in models:
        if model not in param_dict:
            raise ParamRangeConfigError(
                f"Parameter range does not define model '{model}'"
            )
        contents = param_dict[model]
        if not isinstance(contents, dict):
            raise ParamRangeConfigError(
                f"Parameter range for model '{model}' must be a mapping"
            )

        param_names = contents.get("param_name")
        param_ranges = contents.get("param_range")
        if not isinstance(param_names, list) or not param_names:
            raise ParamRangeConfigError(
                f"Model '{model}' must define a non-empty param_name list"
            )
        if not isinstance(param_ranges, dict):
            raise ParamRangeConfigError(
                f"Model '{model}' must define a param_range mapping"
            )

        missing = [name for name in param_names if name not in param_ranges]
        extra = [name for name in param_ranges if name not in param_names]
        if missing:
            raise ParamRangeConfigError(
                f"Model '{model}' is missing ranges for: {missing}"
            )
        if extra:
            raise ParamRangeConfigError(
                f"Model '{model}' has extra ranges not in param_name: {extra}"
            )

        normalized[model] = {
            "param_name": list(param_names),
            "param_range": OrderedDict(
                (
                    name,
                    _validate_param_bounds(model, name, param_ranges[name]),
                )
                for name in param_names
            ),
        }

    return normalized

Parameter Utilities¶

`detect_parameter_format(parameters, param_ranges)` ¶

Detect whether parameters are normalized (0-1 range) or in original scale.

Parameters¶

parameters : np.ndarray Model parameters array [basin, parameter] param_ranges : Dict[str, List[float]] Parameter ranges dictionary with min/max values for each parameter

Returns¶

bool True if parameters appear to be normalized (0-1 range), False otherwise

Source code in hydromodel/models/param_utils.py

def detect_parameter_format(
    parameters: np.ndarray, param_ranges: Dict[str, List[float]]
) -> bool:
    """
    Detect whether parameters are normalized (0-1 range) or in original scale.

    Parameters
    ----------
    parameters : np.ndarray
        Model parameters array [basin, parameter]
    param_ranges : Dict[str, List[float]]
        Parameter ranges dictionary with min/max values for each parameter

    Returns
    -------
    bool
        True if parameters appear to be normalized (0-1 range), False otherwise
    """
    # Check if all parameters are within [0, 1] range
    # Allow small tolerance for numerical precision
    tolerance = 1e-6

    # If any parameter is outside [0, 1] with tolerance, assume original scale
    if np.any(parameters < -tolerance) or np.any(parameters > 1 + tolerance):
        return False

    # Additional check: if parameters are suspiciously close to range boundaries
    # when interpreted as original values, they're likely normalized
    param_values = list(param_ranges.values())
    for i, param_range in enumerate(param_values):
        if i >= parameters.shape[1]:
            break
        param_col = parameters[:, i]
        min_val, max_val = param_range[0], param_range[1]

        # If parameters are all very close to 0-1 range but the actual range
        # is much larger, they're likely normalized
        if max_val - min_val > 2 and np.all(param_col <= 1.1):
            return True

    return True  # Default assumption: parameters are normalized

`get_parameter_scales(param_ranges)` ¶

Extract parameter scales from param_ranges for backward compatibility.

Parameters¶

param_ranges : Dict[str, List[float]] Parameter ranges dictionary

Returns¶

Dict[str, List[float]] Dictionary mapping parameter names to [min, max] ranges

Source code in hydromodel/models/param_utils.py

def get_parameter_scales(
    param_ranges: Dict[str, List[float]],
) -> Dict[str, List[float]]:
    """
    Extract parameter scales from param_ranges for backward compatibility.

    Parameters
    ----------
    param_ranges : Dict[str, List[float]]
        Parameter ranges dictionary

    Returns
    -------
    Dict[str, List[float]]
        Dictionary mapping parameter names to [min, max] ranges
    """
    return param_ranges.copy()

`normalize_parameters(parameters, param_ranges)` ¶

Convert parameters from original scale to normalized (0-1) scale.

Parameters¶

parameters : np.ndarray Model parameters in original scale [basin, parameter] param_ranges : Dict[str, List[float]] Parameter ranges dictionary

Returns¶

np.ndarray Parameters normalized to 0-1 scale

Source code in hydromodel/models/param_utils.py

def normalize_parameters(
    parameters: np.ndarray, param_ranges: Dict[str, List[float]]
) -> np.ndarray:
    """
    Convert parameters from original scale to normalized (0-1) scale.

    Parameters
    ----------
    parameters : np.ndarray
        Model parameters in original scale [basin, parameter]
    param_ranges : Dict[str, List[float]]
        Parameter ranges dictionary

    Returns
    -------
    np.ndarray
        Parameters normalized to 0-1 scale
    """
    normalized_params = np.zeros_like(parameters)
    param_list = list(param_ranges.values())

    for i, param_range in enumerate(param_list):
        min_val, max_val = param_range[0], param_range[1]
        normalized_params[:, i] = (parameters[:, i] - min_val) / (
            max_val - min_val
        )

    return normalized_params

`process_parameters(parameters, param_ranges, normalized='auto')` ¶

Process model parameters to convert from normalized to original scale if needed.

This function provides a unified interface for parameter handling across all models, supporting both normalized (0-1 range) and original scale parameters.

Parameters¶

parameters : np.ndarray Model parameters array [basin, parameter] param_ranges : Dict[str, List[float]] Parameter ranges dictionary with min/max values for each parameter normalized : Union[bool, str], optional Parameter format specification: - "auto": Automatically detect parameter format (default) - True: Parameters are normalized (0-1 range), convert to original scale - False: Parameters are already in original scale, use as-is

Returns¶

np.ndarray Parameters in original scale, ready for model computation

Examples¶

param_ranges = {"K": [0.1, 1.0], "B": [0.1, 0.4]}

Normalized parameters¶

norm_params = np.array([[0.5, 0.8]]) # Will be converted to [0.55, 0.34] orig_params = process_parameters(norm_params, param_ranges, normalized=True)

Original scale parameters¶

orig_params = np.array([[0.55, 0.34]]) # Will be used as-is final_params = process_parameters(orig_params, param_ranges, normalized=False)

Source code in hydromodel/models/param_utils.py

def process_parameters(
    parameters: np.ndarray,
    param_ranges: Dict[str, List[float]],
    normalized: Union[bool, str] = "auto",
) -> np.ndarray:
    """
    Process model parameters to convert from normalized to original scale if needed.

    This function provides a unified interface for parameter handling across all models,
    supporting both normalized (0-1 range) and original scale parameters.

    Parameters
    ----------
    parameters : np.ndarray
        Model parameters array [basin, parameter]
    param_ranges : Dict[str, List[float]]
        Parameter ranges dictionary with min/max values for each parameter
    normalized : Union[bool, str], optional
        Parameter format specification:
        - "auto": Automatically detect parameter format (default)
        - True: Parameters are normalized (0-1 range), convert to original scale
        - False: Parameters are already in original scale, use as-is

    Returns
    -------
    np.ndarray
        Parameters in original scale, ready for model computation

    Examples
    --------
    >>> param_ranges = {"K": [0.1, 1.0], "B": [0.1, 0.4]}
    >>> # Normalized parameters
    >>> norm_params = np.array([[0.5, 0.8]])  # Will be converted to [0.55, 0.34]
    >>> orig_params = process_parameters(norm_params, param_ranges, normalized=True)
    >>>
    >>> # Original scale parameters
    >>> orig_params = np.array([[0.55, 0.34]])  # Will be used as-is
    >>> final_params = process_parameters(orig_params, param_ranges, normalized=False)
    """
    if parameters.shape[1] != len(param_ranges):
        raise ValueError(
            f"Parameter array has {parameters.shape[1]} columns but "
            f"param_ranges has {len(param_ranges)} parameters"
        )

    # Auto-detect parameter format if requested
    if normalized == "auto":
        normalized = detect_parameter_format(parameters, param_ranges)

    # If parameters are already in original scale, return as-is
    if not normalized:
        return parameters.copy()

    # Convert normalized parameters to original scale
    converted_params = np.zeros_like(parameters)
    param_list = list(param_ranges.values())

    for i, param_range in enumerate(param_list):
        min_val, max_val = param_range[0], param_range[1]
        converted_params[:, i] = min_val + parameters[:, i] * (
            max_val - min_val
        )

    return converted_params

`validate_parameters(parameters, param_ranges, normalized=False)` ¶

Validate that parameters are within acceptable ranges.

Parameters¶

parameters : np.ndarray Model parameters array [basin, parameter] param_ranges : Dict[str, List[float]] Parameter ranges dictionary normalized : bool, optional Whether parameters are normalized (default: False)

Returns¶

bool True if all parameters are within valid ranges

Source code in hydromodel/models/param_utils.py

def validate_parameters(
    parameters: np.ndarray,
    param_ranges: Dict[str, List[float]],
    normalized: bool = False,
) -> bool:
    """
    Validate that parameters are within acceptable ranges.

    Parameters
    ----------
    parameters : np.ndarray
        Model parameters array [basin, parameter]
    param_ranges : Dict[str, List[float]]
        Parameter ranges dictionary
    normalized : bool, optional
        Whether parameters are normalized (default: False)

    Returns
    -------
    bool
        True if all parameters are within valid ranges
    """
    if normalized:
        # For normalized parameters, check 0-1 range
        return np.all(parameters >= 0) and np.all(parameters <= 1)
    else:
        # For original scale parameters, check against param_ranges
        param_list = list(param_ranges.values())
        for i, param_range in enumerate(param_list):
            if i >= parameters.shape[1]:
                break
            min_val, max_val = param_range[0], param_range[1]
            param_col = parameters[:, i]
            if np.any(param_col < min_val) or np.any(param_col > max_val):
                return False
        return True

Model Dictionary¶

`check_dependencies()` ¶

Return availability of optional calibration dependencies.

Source code in hydromodel/models/model_dict.py

def check_dependencies():
    """Return availability of optional calibration dependencies."""
    dependencies = {}
    for package in ["deap", "spotpy", "scipy", "xarray"]:
        try:
            __import__(package)
            dependencies[package] = True
        except ImportError:
            dependencies[package] = False
    return dependencies

`describe_model(model_name)` ¶

Return model callable and parameter contract metadata.

Source code in hydromodel/models/model_dict.py

def describe_model(model_name):
    """Return model callable and parameter contract metadata."""
    if model_name not in MODEL_DICT:
        raise KeyError(f"Unsupported model: {model_name}")

    from hydromodel.models.model_config import MODEL_PARAM_DICT

    return {
        "name": model_name,
        "available": True,
        "parameters": MODEL_PARAM_DICT.get(model_name),
    }

`list_losses()` ¶

Return user-facing objectives and registered internal loss keys.

Source code in hydromodel/models/model_dict.py

def list_losses():
    """Return user-facing objectives and registered internal loss keys."""
    return {
        "user_objectives": sorted(_USER_OBJECTIVE_MAP.keys()),
        "registered_losses": sorted(LOSS_DICT.keys()),
        "maximize_losses": sorted(_MAXIMIZE_LOSSES),
    }

`list_models()` ¶

Return registered model names.

Source code in hydromodel/models/model_dict.py

def list_models():
    """Return registered model names."""
    return sorted(MODEL_DICT.keys())

`resolve_loss_config(loss_config)` ¶

Resolve user-facing objective names to minimization loss keys.

Hydromodel optimizers minimize objective values. User-facing metrics such as NSE, KGE, and LogNSE are maximized by mapping them to negated objectives. Existing LOSS_DICT keys remain accepted for compatibility.

Source code in hydromodel/models/model_dict.py

def resolve_loss_config(loss_config):
    """Resolve user-facing objective names to minimization loss keys.

    Hydromodel optimizers minimize objective values. User-facing metrics such as
    NSE, KGE, and LogNSE are maximized by mapping them to negated objectives.
    Existing LOSS_DICT keys remain accepted for compatibility.
    """
    resolved = dict(loss_config or {})
    obj_func = resolved.get("obj_func", "RMSE")

    if callable(obj_func):
        resolved.setdefault(
            "requested_obj_func", getattr(obj_func, "__name__", "callable")
        )
        resolved.setdefault("resolved_obj_func", obj_func)
        return resolved

    requested = str(obj_func)
    requested_upper = requested.upper()

    if requested_upper in _USER_OBJECTIVE_MAP:
        resolved_obj_func = _USER_OBJECTIVE_MAP[requested_upper]
    elif requested in LOSS_DICT:
        resolved_obj_func = requested
        if requested in _MAXIMIZE_LOSSES:
            warnings.warn(
                f"Objective '{requested}' is a higher-is-better metric but "
                "hydromodel optimizers minimize objective values. Use "
                f"'{_MAXIMIZE_LOSS_REPLACEMENTS.get(requested, requested_upper)}' "
                "or the matching negated objective for calibration.",
                RuntimeWarning,
                stacklevel=2,
            )
    elif f"spotpy_{requested.lower()}" in LOSS_DICT:
        resolved_obj_func = f"spotpy_{requested.lower()}"
        if resolved_obj_func in _MAXIMIZE_LOSSES:
            warnings.warn(
                f"Objective '{resolved_obj_func}' is a higher-is-better "
                "metric but hydromodel optimizers minimize objective values. "
                "Use "
                f"'{_MAXIMIZE_LOSS_REPLACEMENTS.get(resolved_obj_func, requested_upper)}' "
                "or the matching negated objective for calibration.",
                RuntimeWarning,
                stacklevel=2,
            )
    else:
        supported = sorted(set(_USER_OBJECTIVE_MAP) | set(LOSS_DICT))
        raise KeyError(
            f"Unsupported objective function '{requested}'. "
            f"Supported values include: {', '.join(supported[:20])}"
        )

    if resolved_obj_func not in LOSS_DICT:
        raise KeyError(
            f"Resolved objective '{resolved_obj_func}' is not registered in LOSS_DICT"
        )

    resolved["requested_obj_func"] = resolved.get(
        "requested_obj_func", requested_upper
    )
    resolved["resolved_obj_func"] = resolved_obj_func
    resolved["obj_func"] = resolved_obj_func
    return resolved

`rmse43darr(obs, sim)` ¶

RMSE for 3D array

Parameters¶

obs : np.ndarray observation data sim : np.ndarray simulation data

Returns¶

type description

Raises¶

ValueError description

Source code in hydromodel/models/model_dict.py

def rmse43darr(obs, sim):
    """RMSE for 3D array

    Parameters
    ----------
    obs : np.ndarray
        observation data
    sim : np.ndarray
        simulation data

    Returns
    -------
    _type_
        _description_

    Raises
    ------
    ValueError
        _description_
    """
    rmses = np.sqrt(np.nanmean((sim - obs) ** 2, axis=0))
    rmse = rmses.mean(axis=0)
    if np.isnan(rmse) or any(np.isnan(sim)):
        raise ValueError(
            "RMSE is nan or there are nan values in the simulation data, "
            "please check the input data."
        )
    # tolist is necessary for spotpy to get the value
    # otherwise the print will incur to an issue
    # https://github.com/thouska/spotpy/issues/319
    return rmse.tolist()

Common Utilities and Configurations¶

Model Configuration¶

ParamRangeConfigError (ValueError) ¶

attach_parameter_contract(result, model_setup) ¶

denormalize_parameter_dict(parameters, model_param_config) ¶

get_model_param_config(model_name, kwargs=None) ¶

inject_model_param_config(model_params, model_name, model_param_config) ¶

resolve_model_param_config(model_name, param_range_file=None, fallback_dir=None, strict=False) ¶

serializable_loss_config(loss_config) ¶

validate_model_param_dict(param_dict, model_name=None) ¶

Parameter Utilities¶

detect_parameter_format(parameters, param_ranges) ¶

Parameters¶

Returns¶

get_parameter_scales(param_ranges) ¶

Parameters¶

Returns¶

normalize_parameters(parameters, param_ranges) ¶

Parameters¶

Returns¶

process_parameters(parameters, param_ranges, normalized='auto') ¶

Parameters¶

Returns¶

Examples¶

Normalized parameters¶

Original scale parameters¶

validate_parameters(parameters, param_ranges, normalized=False) ¶

Parameters¶

Returns¶

Model Dictionary¶

check_dependencies() ¶

describe_model(model_name) ¶

list_losses() ¶

list_models() ¶

resolve_loss_config(loss_config) ¶

rmse43darr(obs, sim) ¶

Parameters¶

Returns¶

Raises¶

`ParamRangeConfigError (ValueError)` ¶

`attach_parameter_contract(result, model_setup)` ¶

`denormalize_parameter_dict(parameters, model_param_config)` ¶

`get_model_param_config(model_name, kwargs=None)` ¶

`inject_model_param_config(model_params, model_name, model_param_config)` ¶

`resolve_model_param_config(model_name, param_range_file=None, fallback_dir=None, strict=False)` ¶

`serializable_loss_config(loss_config)` ¶

`validate_model_param_dict(param_dict, model_name=None)` ¶

`detect_parameter_format(parameters, param_ranges)` ¶

`get_parameter_scales(param_ranges)` ¶

`normalize_parameters(parameters, param_ranges)` ¶

`process_parameters(parameters, param_ranges, normalized='auto')` ¶

`validate_parameters(parameters, param_ranges, normalized=False)` ¶

`check_dependencies()` ¶

`describe_model(model_name)` ¶

`list_losses()` ¶

`list_models()` ¶

`resolve_loss_config(loss_config)` ¶

`rmse43darr(obs, sim)` ¶