Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature/choice unsequa helper #513

Merged
merged 5 commits into from
Jul 8, 2022
Merged

Feature/choice unsequa helper #513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
00db471
Add lists of haz and imp_set
Jul 5, 2022
aac4c9a
Add fraction, add impfset list
Jul 7, 2022
9281241
Update test with list helpers
Jul 7, 2022
d363c7b
Update tutorial for list helpers and haz fraction
Jul 7, 2022
678e335
Merge branch 'develop' into feature/choice-unsequa-helper
Jul 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 94 additions & 39 deletions climada/engine/unsequa/input_var.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,7 @@ def var_to_inputvar(var):
return InputVar(func=lambda: var, distr_dict={})

@staticmethod
def haz(haz, n_ev=None, bounds_int=None, bounds_freq=None):
def haz(haz_list, n_ev=None, bounds_int=None, bounds_frac=None, bounds_freq=None):
"""
Helper wrapper for basic hazard uncertainty input variable

Expand All @@ -234,22 +234,33 @@ def haz(haz, n_ev=None, bounds_int=None, bounds_freq=None):
HI: scale the intensity of all events (homogeneously)
The instensity of all events is multiplied by a number
sampled uniformly from a distribution with (min, max) = bounds_int
HA: scale the fraction of all events (homogeneously)
The fraction of all events is multiplied by a number
sampled uniformly from a distribution with (min, max) = bounds_frac
HF: scale the frequency of all events (homogeneously)
The frequency of all events is multiplied by a number
sampled uniformly from a distribution with (min, max) = bounds_freq
HL: sample uniformly from hazard list
From the provided list of hazard is elements are uniformly
sampled. For example, Hazards outputs from dynamical models
for different input factors.

If a bounds is None, this parameter is assumed to have no uncertainty.

Parameters
----------
haz : climada.hazard.Hazard
The base hazard
haz : List of climada.hazard.Hazard
The list of base hazard. Can be one or many to uniformly sample
from.
n_ev : int, optional
Number of events to be subsampled per sample. Can be equal or
larger than haz.size. The default is None.
bounds_int : (float, float), optional
Bounds of the uniform distribution for the homogeneous intensity
scaling. The default is None.
bounds_frac : (float, float), optional
Bounds of the uniform distribution for the homogeneous fraction
scaling. The default is None.
bounds_freq : (float, float), optional
Bounds of the uniform distribution for the homogeneous frequency
scaling. The default is None.
Expand All @@ -260,16 +271,21 @@ def haz(haz, n_ev=None, bounds_int=None, bounds_freq=None):
Uncertainty input variable for a hazard object.

"""
kwargs = {'haz': haz, 'n_ev': n_ev}
n_haz = len(haz_list)
kwargs = {'haz_list': haz_list, 'n_ev': n_ev}
if n_ev is None:
kwargs['HE'] = None
if bounds_int is None:
kwargs['HI'] = None
if bounds_frac is None:
kwargs['HA'] = None
if bounds_freq is None:
kwargs['HF'] = None
if n_haz == 1:
kwargs['HL'] = 0
return InputVar(
partial(_haz_uncfunc, **kwargs),
_haz_unc_dict(n_ev, bounds_int, bounds_freq)
_haz_unc_dict(n_ev, bounds_int, bounds_frac, bounds_freq, n_haz)
)

@staticmethod
Expand Down Expand Up @@ -327,7 +343,7 @@ def exp(exp_list, bounds_totval=None, bounds_noise=None):
)

@staticmethod
def impfset(impf_set, haz_id_dict= None, bounds_mdd=None, bounds_paa=None,
def impfset(impf_set_list, haz_id_dict= None, bounds_mdd=None, bounds_paa=None,
bounds_impfi=None):
"""
Helper wrapper for basic impact function set uncertainty input variable.
Expand All @@ -347,13 +363,20 @@ def impfset(impf_set, haz_id_dict= None, bounds_mdd=None, bounds_paa=None,
The value intensity are all summed with a random number
sampled uniformly from a distribution with
(min, max) = bounds_int
IL: sample uniformly from impact function set list
From the provided list of impact function sets elements are uniformly
sampled. For example, impact functions obtained from different
calibration methods.


If a bounds is None, this parameter is assumed to have no uncertainty.

Parameters
----------
impf_set : climada.entity.impact_funcs.impact_func_set.ImpactFuncSet
The base impact function set.
impf_set_list : list of ImpactFuncSet
The list of base impact function set. Can be one or many to
uniformly sample from. The impact function ids must identical
for all impact function sets.
bounds_mdd : (float, float), optional
Bounds of the uniform distribution for the homogeneous mdd
scaling. The default is None.
Expand All @@ -375,25 +398,28 @@ def impfset(impf_set, haz_id_dict= None, bounds_mdd=None, bounds_paa=None,
Uncertainty input variable for an impact function set object.

"""
kwargs = {}
n_impf_set = len(impf_set_list)
kwargs = {'impf_set_list': impf_set_list}
if bounds_mdd is None:
kwargs['MDD'] = None
if bounds_paa is None:
kwargs['PAA'] = None
if bounds_impfi is None:
kwargs['IFi'] = None
if haz_id_dict is None:
haz_id_dict = impf_set.get_ids()
haz_id_dict = impf_set_list[0].get_ids()
if n_impf_set == 1:
kwargs['IL'] = 0
return InputVar(
partial(
_impfset_uncfunc, impf_set=impf_set, haz_id_dict=haz_id_dict,
_impfset_uncfunc, haz_id_dict=haz_id_dict,
**kwargs
),
_impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa)
_impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa, n_impf_set)
)

@staticmethod
def ent(impf_set, disc_rate, exp_list, meas_set, haz_id_dict,
def ent(impf_set_list, disc_rate, exp_list, meas_set, haz_id_dict,
bounds_disc=None, bounds_cost=None, bounds_totval=None,
bounds_noise=None, bounds_mdd=None, bounds_paa=None,
bounds_impfi=None):
Expand Down Expand Up @@ -436,6 +462,10 @@ def ent(impf_set, disc_rate, exp_list, meas_set, haz_id_dict,
The value intensity are all summed with a random number
sampled uniformly from a distribution with
(min, max) = bounds_int
IL: sample uniformly from impact function set list
From the provided list of impact function sets elements are uniformly
sampled. For example, impact functions obtained from different
calibration methods.


If a bounds is None, this parameter is assumed to have no uncertainty.
Expand Down Expand Up @@ -464,8 +494,10 @@ def ent(impf_set, disc_rate, exp_list, meas_set, haz_id_dict,
bounds_impfi : (float, float), optional
Bounds of the uniform distribution for the homogeneous shift
of intensity. The default is None.
impf_set : climada.engine.impact_funcs.impact_func_set.ImpactFuncSet
The base impact function set.
impf_set_list : list of ImpactFuncSet
The list of base impact function set. Can be one or many to
uniformly sample from. The impact function ids must identical
for all impact function sets.
disc_rate : climada.entity.disc_rates.base.DiscRates
The base discount rates.
exp_list : [climada.entity.exposures.base.Exposure]
Expand All @@ -485,6 +517,7 @@ def ent(impf_set, disc_rate, exp_list, meas_set, haz_id_dict,

"""
n_exp = len(exp_list)
n_impf_set = len(impf_set_list)

kwargs = {}
if bounds_mdd is None:
Expand All @@ -493,6 +526,8 @@ def ent(impf_set, disc_rate, exp_list, meas_set, haz_id_dict,
kwargs['PAA'] = None
if bounds_impfi is None:
kwargs['IFi'] = None
if n_impf_set== 1:
kwargs['IL'] = 0
if bounds_disc is None:
kwargs['DR'] = None
if bounds_cost is None:
Expand All @@ -506,18 +541,19 @@ def ent(impf_set, disc_rate, exp_list, meas_set, haz_id_dict,

return InputVar(
partial(_ent_unc_func,
impf_set=impf_set, haz_id_dict=haz_id_dict,
impf_set_list=impf_set_list, haz_id_dict=haz_id_dict,
disc_rate=disc_rate, bounds_noise=bounds_noise,
exp_list=exp_list, meas_set=meas_set, **kwargs
),
_ent_unc_dict(bounds_totval=bounds_totval, bounds_noise=bounds_noise,
bounds_impfi=bounds_impfi, bounds_mdd=bounds_mdd,
bounds_impfi=bounds_impfi, n_impf_set=n_impf_set,
bounds_mdd=bounds_mdd,
bounds_paa=bounds_paa, bounds_disc=bounds_disc,
bounds_cost=bounds_cost, n_exp=n_exp)
bounds_cost=bounds_cost, n_exp=n_exp,)
)

@staticmethod
def entfut(impf_set, exp_list, meas_set, haz_id_dict,
def entfut(impf_set_list, exp_list, meas_set, haz_id_dict,
bounds_cost=None, bounds_eg=None, bounds_noise=None,
bounds_impfi=None, bounds_mdd=None, bounds_paa=None,
):
Expand Down Expand Up @@ -556,6 +592,10 @@ def entfut(impf_set, exp_list, meas_set, haz_id_dict,
The value intensity are all summed with a random number
sampled uniformly from a distribution with
(min, max) = bounds_impfi
IL: sample uniformly from impact function set list
From the provided list of impact function sets elements are uniformly
sampled. For example, impact functions obtained from different
calibration methods.


If a bounds is None, this parameter is assumed to have no uncertainty.
Expand All @@ -581,8 +621,10 @@ def entfut(impf_set, exp_list, meas_set, haz_id_dict,
bounds_impfi : (float, float), optional
Bounds of the uniform distribution for the homogeneous shift
of intensity. The default is None.
impf_set : climada.engine.impact_funcs.impact_func_set.ImpactFuncSet
The base impact function set.
impf_set_list : list of ImpactFuncSet
The list of base impact function set. Can be one or many to
uniformly sample from. The impact function ids must identical
for all impact function sets.
exp_list : [climada.entity.exposures.base.Exposure]
The list of base exposure. Can be one or many to uniformly sample
from.
Expand All @@ -600,6 +642,7 @@ def entfut(impf_set, exp_list, meas_set, haz_id_dict,

"""
n_exp = len(exp_list)
n_impf_set = len(impf_set_list)

kwargs = {}
if bounds_mdd is None:
Expand All @@ -608,6 +651,8 @@ def entfut(impf_set, exp_list, meas_set, haz_id_dict,
kwargs['PAA'] = None
if bounds_impfi is None:
kwargs['IFi'] = None
if n_impf_set == 1:
kwargs['IL'] = 0
if bounds_cost is None:
kwargs['CO'] = None
if bounds_eg is None:
Expand All @@ -619,38 +664,46 @@ def entfut(impf_set, exp_list, meas_set, haz_id_dict,

return InputVar(
partial(_entfut_unc_func, haz_id_dict=haz_id_dict,
bounds_noise=bounds_noise, impf_set=impf_set,
bounds_noise=bounds_noise, impf_set_list=impf_set_list,
exp_list=exp_list, meas_set=meas_set, **kwargs),
_entfut_unc_dict(bounds_eg=bounds_eg, bounds_noise=bounds_noise,
bounds_impfi=bounds_impfi, bounds_paa=bounds_paa,
bounds_impfi=bounds_impfi, n_impf_set=n_impf_set,
bounds_paa=bounds_paa,
bounds_mdd=bounds_mdd, bounds_cost=bounds_cost,
n_exp=n_exp)
)


#Hazard
def _haz_uncfunc(HE, HI, HF, haz, n_ev):
haz_tmp = copy.deepcopy(haz)
def _haz_uncfunc(HE, HI, HA, HF, HL, haz_list, n_ev):
haz_tmp = copy.deepcopy(haz_list[int(HL)])
if HE is not None:
rng = np.random.RandomState(int(HE))
event_id = list(rng.choice(haz_tmp.event_id, int(n_ev)))
haz_tmp = haz_tmp.select(event_id=event_id)
if HI is not None:
haz_tmp.intensity = haz_tmp.intensity.multiply(HI)
if HA is not None:
haz_tmp.fraction = haz_tmp.fraction.multiply(HA)
if HF is not None:
haz_tmp.frequency = np.multiply(haz_tmp.frequency, HF)
return haz_tmp

def _haz_unc_dict(n_ev, bounds_int, bounds_freq):
def _haz_unc_dict(n_ev, bounds_int, bounds_frac, bounds_freq, n_haz):
hud = {}
if n_ev is not None:
hud['HE'] = sp.stats.randint(0, 2**32 - 1) #seed for rnd generator
if bounds_int is not None:
imin, idelta = bounds_int[0], bounds_int[1] - bounds_int[0]
hud['HI'] = sp.stats.uniform(imin, idelta)
if bounds_frac is not None:
amin, adelta = bounds_frac[0], bounds_frac[1] - bounds_frac[0]
hud['HA'] = sp.stats.uniform(amin, adelta)
if bounds_freq is not None:
fmin, fdelta = bounds_freq[0], bounds_freq[1] - bounds_freq[0]
hud['HF'] = sp.stats.uniform(fmin, fdelta)
if n_haz > 1:
hud['HL'] = sp.stats.randint(0, n_haz)
return hud

#Exposure
Expand All @@ -676,8 +729,8 @@ def _exp_unc_dict(bounds_totval, bounds_noise, n_exp):
return eud

#Impact function set
def _impfset_uncfunc(IFi, MDD, PAA, impf_set, haz_id_dict):
impf_set_tmp = copy.deepcopy(impf_set)
def _impfset_uncfunc(IFi, MDD, PAA, IL, impf_set_list, haz_id_dict):
impf_set_tmp = copy.deepcopy(impf_set_list[int(IL)])
for haz_type, fun_id_list in haz_id_dict.items():
for fun_id in fun_id_list:
if MDD is not None:
Expand All @@ -700,7 +753,7 @@ def _impfset_uncfunc(IFi, MDD, PAA, impf_set, haz_id_dict):
impf_set_tmp.get_func(haz_type=haz_type, fun_id=fun_id).intensity = new_int
return impf_set_tmp

def _impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa):
def _impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa, n_impf_set):
iud = {}
if bounds_impfi is not None:
xmin, xdelta = bounds_impfi[0], bounds_impfi[1] - bounds_impfi[0]
Expand All @@ -711,6 +764,8 @@ def _impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa):
if bounds_mdd is not None:
xmin, xdelta = bounds_mdd[0], bounds_mdd[1] - bounds_mdd[0]
iud['MDD'] = sp.stats.uniform(xmin, xdelta)
if n_impf_set > 1:
iud['IL'] = sp.stats.randint(0, n_impf_set)
return iud

#Entity
Expand Down Expand Up @@ -738,36 +793,36 @@ def _meas_set_unc_dict(bounds_cost):
cmin, cdelta = bounds_cost[0], bounds_cost[1] - bounds_cost[0]
return {'CO': sp.stats.uniform(cmin, cdelta)}

def _ent_unc_func(EN, ET, EL, IFi, MDD, PAA, CO, DR, bounds_noise,
impf_set, haz_id_dict, disc_rate, exp_list, meas_set):
def _ent_unc_func(EN, ET, EL, IFi, IL, MDD, PAA, CO, DR, bounds_noise,
impf_set_list, haz_id_dict, disc_rate, exp_list, meas_set):
ent = Entity()
ent.exposures = _exp_uncfunc(EN, ET, EL, exp_list, bounds_noise)
ent.impact_funcs = _impfset_uncfunc(IFi, MDD, PAA, impf_set=impf_set,
ent.impact_funcs = _impfset_uncfunc(IFi, MDD, PAA, IL, impf_set_list=impf_set_list,
haz_id_dict=haz_id_dict)
ent.measures = _meas_set_uncfunc(CO, meas_set=meas_set)
ent.disc_rates = _disc_uncfunc(DR, disc_rate)
return ent

def _ent_unc_dict(bounds_totval, bounds_noise, bounds_impfi, bounds_mdd,
bounds_paa, bounds_disc, bounds_cost, n_exp):
bounds_paa, n_impf_set, bounds_disc, bounds_cost, n_exp):
ent_unc_dict = _exp_unc_dict(bounds_totval, bounds_noise, n_exp)
ent_unc_dict.update(_impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa))
ent_unc_dict.update(_impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa, n_impf_set))
ent_unc_dict.update(_disc_unc_dict(bounds_disc))
ent_unc_dict.update(_meas_set_unc_dict(bounds_cost))
return ent_unc_dict

def _entfut_unc_func(EN, EG, EL, IFi, MDD, PAA, CO, bounds_noise,
impf_set, haz_id_dict, exp_list, meas_set):
def _entfut_unc_func(EN, EG, EL, IFi, IL, MDD, PAA, CO, bounds_noise,
impf_set_list, haz_id_dict, exp_list, meas_set):
ent = Entity()
ent.exposures = _exp_uncfunc(EN=EN, ET=EG, EL=EL, exp_list=exp_list, bounds_noise=bounds_noise)
ent.impact_funcs = _impfset_uncfunc(IFi, MDD, PAA, impf_set=impf_set,
ent.impact_funcs = _impfset_uncfunc(IFi, MDD, PAA, IL, impf_set_list=impf_set_list,
haz_id_dict=haz_id_dict)
ent.measures = _meas_set_uncfunc(CO, meas_set=meas_set)
ent.disc_rates = DiscRates() #Disc rate of future entity ignored in cost_benefit.calc()
return ent

def _entfut_unc_dict(bounds_impfi, bounds_mdd,
bounds_paa, bounds_eg, bounds_noise,
bounds_paa, n_impf_set, bounds_eg, bounds_noise,
bounds_cost, n_exp):
eud = {}
if bounds_eg is not None:
Expand All @@ -777,7 +832,7 @@ def _entfut_unc_dict(bounds_impfi, bounds_mdd,
eud['EN'] = sp.stats.randint(0, 2**32 - 1) #seed for rnd generator
if n_exp > 1:
eud['EL'] = sp.stats.randint(0, n_exp)
eud.update(_impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa))
eud.update(_impfset_unc_dict(bounds_impfi, bounds_mdd, bounds_paa, n_impf_set))
if bounds_cost is not None:
eud.update(_meas_set_unc_dict(bounds_cost))
return eud
4 changes: 2 additions & 2 deletions climada/engine/unsequa/test/test_unsequa.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ def make_costben_iv():

entdem = ent_dem()
ent_iv = InputVar.ent(
impf_set = entdem.impact_funcs,
impf_set_list = [entdem.impact_funcs],
disc_rate = entdem.disc_rates,
exp_list = [entdem.exposures],
meas_set = entdem.measures,
Expand All @@ -134,7 +134,7 @@ def make_costben_iv():

entfutdem = ent_fut_dem()
entfut_iv = InputVar.entfut(
impf_set = entfutdem.impact_funcs,
impf_set_list = [entfutdem.impact_funcs],
exp_list = [entfutdem.exposures],
meas_set = entfutdem.measures,
bounds_eg=[0.8, 1.5],
Expand Down
Loading