Skip to content

Commit

Permalink
Feature/choice unsequa helper (#513)
Browse files Browse the repository at this point in the history 
* Add lists of haz and imp_set

* Add fraction, add impfset list

* Update test with list helpers

* Update tutorial for list helpers and haz fraction

Co-authored-by: Chahan Kropf <[email protected]>
  • Loading branch information
@chahank
chahank and Chahan Kropf authored Jul 8, 2022
1 parent 42245f2 commit 63b5fb7
Show file tree
Hide file tree
Showing 3 changed files with 1,049 additions and 383 deletions.
133 changes: 94 additions & 39 deletions climada/engine/unsequa/input_var.py
View file
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
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

0 comments on commit 63b5fb7

Please sign in to comment.