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.

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/init measures base #564

Merged
merged 9 commits into from
Oct 28, 2022
Merged
90 changes: 73 additions & 17 deletions climada/entity/measures/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import copy
import logging
from pathlib import Path
from typing import Optional, Tuple

import numpy as np
import pandas as pd
from geopandas import GeoDataFrame
Expand Down Expand Up @@ -81,33 +83,87 @@ class Measure():
cost of risk transfer
"""

def __init__(self):
"""Empty initialization."""
self.name = ''
self.haz_type = ''
self.color_rgb = np.array([0, 0, 0])
self.cost = 0
def __init__(
self,
name: str = "",
haz_type: str = "",
cost: float = 0,
hazard_set: str = NULL_STR,
hazard_freq_cutoff: float = 0,
exposures_set: str = NULL_STR,
imp_fun_map: str = NULL_STR,
hazard_inten_imp: Tuple[float, float] = (1, 0),
mdd_impact: Tuple[float, float] = (1, 0),
paa_impact: Tuple[float, float] = (1, 0),
exp_region_id: Optional[list] = None,
risk_transf_attach: float = 0,
risk_transf_cover: float = 0,
risk_transf_cost_factor: float = 1,
color_rgb: Optional[np.ndarray] = None
):
"""Initialize a Measure object with given values.

Parameters
----------
name : str, optional
name of the measure
haz_type : str, optional
related hazard type (peril), e.g. TC
cost : float, optional
discounted cost (in same units as assets)
hazard_set : str, optional
file name of hazard to use (in h5 format)
hazard_freq_cutoff : float, optional
hazard frequency cutoff
exposures_set : str or climada.entity.Exposure, optional
file name of exposure to use (in h5 format) or Exposure instance
imp_fun_map : str, optional
change of impact function id of exposures, e.g. '1to3'
hazard_inten_imp : tuple(float, float), optional
parameter a and b of hazard intensity change
mdd_impact : tuple(float, float), optional
parameter a and b of the impact over the mean damage degree
paa_impact : tuple(float, float), optional
parameter a and b of the impact over the percentage of affected assets
exp_region_id : int, optional
region id of the selected exposures to consider ALL the previous
parameters
risk_transf_attach : float, optional
risk transfer attachment
risk_transf_cover : float, optional
risk transfer cover
risk_transf_cost_factor : float, optional
factor to multiply to resulting insurance layer to get the total
cost of risk transfer
color_rgb : np.array, optional
integer array of size 3. Color code of this measure in RGB.
Default is None (corresponds to black).
"""
self.name = name
self.haz_type = haz_type
self.color_rgb = np.array([0, 0, 0]) if color_rgb is None else color_rgb
self.cost = cost

# related to change in hazard
self.hazard_set = NULL_STR
self.hazard_freq_cutoff = 0
self.hazard_set = hazard_set
self.hazard_freq_cutoff = hazard_freq_cutoff

# related to change in exposures
self.exposures_set = NULL_STR
self.imp_fun_map = NULL_STR # ids of impact functions to change e.g. 1to10
self.exposures_set = exposures_set
self.imp_fun_map = imp_fun_map

# related to change in impact functions
self.hazard_inten_imp = (1, 0) # parameter a and b
self.mdd_impact = (1, 0) # parameter a and b
self.paa_impact = (1, 0) # parameter a and b
self.hazard_inten_imp = hazard_inten_imp
self.mdd_impact = mdd_impact
self.paa_impact = paa_impact

# related to change in region
self.exp_region_id = []
self.exp_region_id = [] if exp_region_id is None else exp_region_id

# risk transfer
self.risk_transf_attach = 0
self.risk_transf_cover = 0
self.risk_transf_cost_factor = 1
self.risk_transf_attach = risk_transf_attach
self.risk_transf_cover = risk_transf_cover
self.risk_transf_cost_factor = risk_transf_cost_factor

def check(self):
"""
Expand Down
127 changes: 69 additions & 58 deletions climada/entity/measures/measure_set.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,12 +101,13 @@ def __init__(self):
--------
Fill MeasureSet with values and check consistency data:

>>> act_1 = Measure()
>>> act_1.name = 'Seawall'
>>> act_1.color_rgb = np.array([0.1529, 0.2510, 0.5451])
>>> act_1.hazard_intensity = (1, 0)
>>> act_1.mdd_impact = (1, 0)
>>> act_1.paa_impact = (1, 0)
>>> act_1 = Measure(
... name='Seawall',
... color_rgb=np.array([0.1529, 0.2510, 0.5451]),
... hazard_intensity=(1, 0),
... mdd_impact=(1, 0),
... paa_impact=(1, 0),
... )
>>> meas = MeasureSet()
>>> meas.append(act_1)
>>> meas.tag.description = "my dummy MeasureSet."
Expand Down Expand Up @@ -357,45 +358,46 @@ def read_att_mat(measures, data, file_name, var_names):
"""Read MATLAB measures attributes"""
num_mes = len(data[var_names['var_name']['name']])
for idx in range(0, num_mes):
meas = Measure()

meas.name = u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['name']][idx][0])

color_str = u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['color']][idx][0])
meas.color_rgb = np.fromstring(color_str, dtype=float, sep=' ')
meas.cost = data[var_names['var_name']['cost']][idx][0]
meas.haz_type = u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['haz']][idx][0])
meas.hazard_freq_cutoff = data[var_names['var_name']['haz_frq']][idx][0]
meas.hazard_set = u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['haz_set']][idx][0])

try:
meas.hazard_inten_imp = (
hazard_inten_imp = (
data[var_names['var_name']['haz_int_a']][idx][0],
data[var_names['var_name']['haz_int_b']][0][idx])
except KeyError:
meas.hazard_inten_imp = (
hazard_inten_imp = (
data[var_names['var_name']['haz_int_a'][:-2]][idx][0], 0)

# different convention of signes followed in MATLAB!
meas.mdd_impact = (data[var_names['var_name']['mdd_a']][idx][0],
data[var_names['var_name']['mdd_b']][idx][0])
meas.paa_impact = (data[var_names['var_name']['paa_a']][idx][0],
data[var_names['var_name']['paa_b']][idx][0])
meas.imp_fun_map = u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['fun_map']][idx][0])
meas_kwargs = dict(
name=u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['name']][idx][0]),
color_rgb=np.fromstring(color_str, dtype=float, sep=' '),
cost=data[var_names['var_name']['cost']][idx][0],
haz_type=u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['haz']][idx][0]),
hazard_freq_cutoff=data[var_names['var_name']['haz_frq']][idx][0],
hazard_set=u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['haz_set']][idx][0]),
hazard_inten_imp=hazard_inten_imp,
# different convention of signs followed in MATLAB!
mdd_impact=(data[var_names['var_name']['mdd_a']][idx][0],
data[var_names['var_name']['mdd_b']][idx][0]),
paa_impact=(data[var_names['var_name']['paa_a']][idx][0],
data[var_names['var_name']['paa_b']][idx][0]),
imp_fun_map=u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['fun_map']][idx][0]),
exposures_set=u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['exp_set']][idx][0]),
risk_transf_attach=data[var_names['var_name']['risk_att']][idx][0],
risk_transf_cover=data[var_names['var_name']['risk_cov']][idx][0],
)

meas.exposures_set = u_hdf5.get_str_from_ref(
file_name, data[var_names['var_name']['exp_set']][idx][0])
exp_region_id = data[var_names['var_name']['exp_reg']][idx][0]
if exp_region_id:
meas.exp_region_id = [exp_region_id]
meas.risk_transf_attach = data[var_names['var_name']['risk_att']][idx][0]
meas.risk_transf_cover = data[var_names['var_name']['risk_cov']][idx][0]
meas_kwargs["exp_region_id"] = [exp_region_id]

measures.append(meas)
measures.append(Measure(**meas_kwargs))

data = u_hdf5.read(file_name)
meas_set = cls()
Expand Down Expand Up @@ -444,47 +446,56 @@ def read_att_excel(measures, dfr, var_names):
"""Read Excel measures attributes"""
num_mes = len(dfr.index)
for idx in range(0, num_mes):
meas = Measure()
# Search for (a, b) values, put a=1 otherwise
try:
hazard_inten_imp = (dfr[var_names['col_name']['haz_int_a']][idx],
dfr[var_names['col_name']['haz_int_b']][idx])
except KeyError:
hazard_inten_imp = (1, dfr['hazard intensity impact'][idx])

meas_kwargs = dict(
name=dfr[var_names['col_name']['name']][idx],
cost=dfr[var_names['col_name']['cost']][idx],
hazard_freq_cutoff=dfr[var_names['col_name']['haz_frq']][idx],
hazard_set=dfr[var_names['col_name']['haz_set']][idx],
hazard_inten_imp=hazard_inten_imp,
mdd_impact=(dfr[var_names['col_name']['mdd_a']][idx],
dfr[var_names['col_name']['mdd_b']][idx]),
paa_impact=(dfr[var_names['col_name']['paa_a']][idx],
dfr[var_names['col_name']['paa_b']][idx]),
imp_fun_map=dfr[var_names['col_name']['fun_map']][idx],
risk_transf_attach=dfr[var_names['col_name']['risk_att']][idx],
risk_transf_cover=dfr[var_names['col_name']['risk_cov']][idx],
color_rgb=np.fromstring(
dfr[var_names['col_name']['color']][idx], dtype=float, sep=' '),
)

meas.name = dfr[var_names['col_name']['name']][idx]
try:
meas.haz_type = dfr[var_names['col_name']['haz']][idx]
meas_kwargs["haz_type"] = dfr[var_names['col_name']['haz']][idx]
except KeyError:
pass
meas.color_rgb = np.fromstring(
dfr[var_names['col_name']['color']][idx], dtype=float, sep=' ')
meas.cost = dfr[var_names['col_name']['cost']][idx]

meas.hazard_freq_cutoff = dfr[var_names['col_name']['haz_frq']][idx]
meas.hazard_set = dfr[var_names['col_name']['haz_set']][idx]
# Search for (a, b) values, put a = 1 otherwise
try:
meas.hazard_inten_imp = (dfr[var_names['col_name']['haz_int_a']][idx],
dfr[var_names['col_name']['haz_int_b']][idx])
meas_kwargs["exposures_set"] = dfr[var_names['col_name']['exp_set']][idx]
except KeyError:
meas.hazard_inten_imp = (1, dfr['hazard intensity impact'][idx])
pass

try:
meas.exposures_set = dfr[var_names['col_name']['exp_set']][idx]
meas.exp_region_id = ast.literal_eval(dfr[var_names['col_name']['exp_reg']][idx])
meas_kwargs["exp_region_id"] = ast.literal_eval(
dfr[var_names['col_name']['exp_reg']][idx])
except KeyError:
pass
except ValueError:
meas.exp_region_id = dfr[var_names['col_name']['exp_reg']][idx]

meas.mdd_impact = (dfr[var_names['col_name']['mdd_a']][idx],
dfr[var_names['col_name']['mdd_b']][idx])
meas.paa_impact = (dfr[var_names['col_name']['paa_a']][idx],
dfr[var_names['col_name']['paa_b']][idx])
meas.imp_fun_map = dfr[var_names['col_name']['fun_map']][idx]
meas.risk_transf_attach = dfr[var_names['col_name']['risk_att']][idx]
meas.risk_transf_cover = dfr[var_names['col_name']['risk_cov']][idx]
meas_kwargs["exp_region_id"] = dfr[var_names['col_name']['exp_reg']][idx]

try:
meas.risk_transf_cost_factor = dfr[var_names['col_name']['risk_fact']][idx]
meas_kwargs["risk_transf_cost_factor"] = (
dfr[var_names['col_name']['risk_fact']][idx]
)
except KeyError:
pass

measures.append(meas)
measures.append(Measure(**meas_kwargs))

dfr = pd.read_excel(file_name, var_names['sheet_name'])
dfr = dfr.fillna('')
Expand Down
23 changes: 11 additions & 12 deletions climada/entity/measures/test/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -144,9 +144,10 @@ def test_cutoff_hazard_region_pass(self):

def test_change_exposures_impf_pass(self):
"""Test _change_exposures_impf"""
meas = Measure()
meas.imp_fun_map = '1to3'
meas.haz_type = 'TC'
meas = Measure(
imp_fun_map='1to3',
haz_type='TC',
)

imp_set = ImpactFuncSet()
imp_tc = ImpactFunc()
Expand Down Expand Up @@ -179,8 +180,7 @@ def test_change_exposures_impf_pass(self):

def test_change_all_hazard_pass(self):
"""Test _change_all_hazard method"""
meas = Measure()
meas.hazard_set = HAZ_DEMO_H5
meas = Measure(hazard_set=HAZ_DEMO_H5)

ref_haz = Hazard.from_hdf5(HAZ_DEMO_H5)

Expand All @@ -198,8 +198,7 @@ def test_change_all_hazard_pass(self):

def test_change_all_exposures_pass(self):
"""Test _change_all_exposures method"""
meas = Measure()
meas.exposures_set = EXP_DEMO_H5
meas = Measure(exposures_set=EXP_DEMO_H5)

ref_exp = Exposures.from_hdf5(EXP_DEMO_H5)

Expand All @@ -218,8 +217,7 @@ def test_change_all_exposures_pass(self):

def test_not_filter_exposures_pass(self):
"""Test _filter_exposures method with []"""
meas = Measure()
meas.exp_region_id = []
meas = Measure(exp_region_id=[])

exp = Exposures()
imp_set = ImpactFuncSet()
Expand All @@ -242,9 +240,10 @@ def test_not_filter_exposures_pass(self):

def test_filter_exposures_pass(self):
"""Test _filter_exposures method with two values"""
meas = Measure()
meas.exp_region_id = [3, 4]
meas.haz_type = 'TC'
meas = Measure(
exp_region_id=[3, 4],
haz_type='TC',
)

exp = Exposures.from_mat(ENT_TEST_MAT)
exp.gdf.rename(columns={'impf_': 'impf_TC', 'centr_': 'centr_TC'}, inplace=True)
Expand Down
Loading