# SPDX-License-Identifier: AGPL-3.0-or-later
# Copyright (C) 2025-2026 Djura | Risk - Data - Engineering S.r.l.
from typing import Union, List
import numpy as np
from scipy.stats import norm
from ..utilities import ( # noqa: F401 (re-exported)
to_json_serializable, get_func_args, filter_args, read_json)
[docs]
def cdf_lognormal(xs: Union[List, np.ndarray],
median: float, beta: float) -> np.ndarray:
# Equivalent to:
# prob = stats.norm.cdf((np.log(xs / median)) / beta)
from scipy.stats import lognorm
prob = lognorm.cdf(xs, beta, scale=np.exp(median))
return prob
[docs]
def uncensored_regression(pars: List[float], x: np.ndarray, y: np.ndarray):
"""Uncensored regression to estimate the expected EDP given IM
and the respective uncertainty due to record-to-record variability
Parameters
----------
pars : List[float]
Fitting function parameters
x : np.ndarray
For example, Intensity measure [im] values in m/s
y : np.ndarray
For example, Engineering demand parameters (EDPs)
Returns
-------
float
loss value to minimize
"""
from scipy.stats import norm
mean_ln = pars[1] + pars[0] * x
var = pars[2]
p = norm.pdf(y, loc=mean_ln, scale=var)
return -np.sum(np.log(p[p != 0]))
[docs]
def censored_regression(
pars: List[float], x_unc: np.ndarray, x_cens: np.ndarray,
y_unc: np.ndarray, y_cens: float):
"""Censored regression to estimate the expected EDP given IM
and the respective uncertainty due to record-to-record variability
Parameters
----------
pars : List[float]
Fitting function parameters
x_unc : np.ndarray
For example, uncensored Intensity measure [im] values in m/s
x_cens : np.ndarray
For example, censored Intensity measure [im] values in m/s
y_unc : np.ndarray
For example, uncensored Engineering demand parameters (EDPs)
y_cens : float
For example, censored Engineering demand parameters (EDPs)
Returns
-------
float
loss value to minimize
"""
from scipy.stats import norm
mean_ln_unc = pars[1] + pars[0] * x_unc
mean_ln_cens = pars[1] + pars[0] * x_cens
var = pars[2]
p_unc = norm.pdf(y_unc, loc=mean_ln_unc, scale=var)
p_cens = 1 - norm.cdf(y_cens, loc=mean_ln_cens, scale=var)
return -np.sum(np.log(p_unc[p_unc != 0])) \
- np.sum(np.log(p_cens[p_cens != 0]))
[docs]
def residual_log_dist(pars: list, x: np.ndarray, y: np.ndarray):
y_pred = cdf_lognormal(x, pars[0], pars[1])
return abs(y - y_pred)
[docs]
def get_loss_uncertainties(loss: np.ndarray) -> np.ndarray:
"""Estimate uncertainties
References
----------
Silva, V. (2019) Uncertainty and correlation in seismic vulnerability
functions of building classes. Earthquake Spectra.
DOI: 10.1193/013018eqs031m.
Parameters
----------
loss : np.ndarray
Loss values
Returns
----------
np.ndarray
Uncertainties
"""
loss = loss / np.max(loss)
uncertainties = np.zeros(loss.shape)
loss_filter = (loss > 0.0) & (loss < 1.0)
loss_filtered = loss[loss_filter]
uncertainties[loss_filter] = np.sqrt(loss_filtered * (
-0.7 - 2 * loss_filtered + np.sqrt(6.8 * loss_filtered + 0.5)))
return uncertainties
[docs]
def get_cdf(theta, beta, iml_range=None):
if iml_range is None:
iml_range = np.linspace(0, 5, 50)
probs = norm.cdf(
np.log(iml_range / theta) / beta, loc=0, scale=1
)
return iml_range, probs
DUCTILITY_F = 12.0
[docs]
def set_ductilities(ductility_f):
return np.arange(0.01, ductility_f, 0.01)
[docs]
def deep_merge(dict1, dict2):
"""Recursively merge dict2 into dict1"""
result = dict1.copy()
for key, value in dict2.items():
if key in result and isinstance(result[key], dict) \
and isinstance(value, dict):
result[key] = deep_merge(result[key], value)
else:
result[key] = value
return result