Skip to content

Commit

Permalink
Feature/init measures base (#564)
Browse files Browse the repository at this point in the history 
Refactor Measure.__init__ and use refactored constructored in tutorials and other parts of the CLIMADA core code

Co-authored-by: Chahan M. Kropf <[email protected]>
  • Loading branch information
@tovogt @chahank
tovogt and chahank authored Oct 28, 2022
1 parent 06a35a1 commit 0a7caf9
Show file tree
Hide file tree
Showing 7 changed files with 400 additions and 315 deletions.
90 changes: 73 additions & 17 deletions climada/entity/measures/base.py
View file
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
View file
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
View file
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()

Expand Down Expand Up @@ -175,8 +176,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 @@ -194,8 +194,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 @@ -214,8 +213,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 @@ -238,9 +236,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

0 comments on commit 0a7caf9

Please sign in to comment.