Skip to content

Commit

Permalink
Set all ImpactFunc attributes in __init__ (#550)
Browse files Browse the repository at this point in the history
* Add all ImpactFunc attributes to ImpactFunc.__init__

* Add type hints to ImpactFunc.__init__

* Update ImpactFunc classmethods to new init

* Update tests to new ImpactFunc.__init__

* Update docs on InputVar regarding ImpactFunc

* Update tutorials with new ImpactFunc.__init__

* Update ImpactFunc.__init__ docstring

* Update argument order of ImpactFunc.__init__

* Fix linter issue

* Update tests to new ImpactFunc.__init__

* Update tutorials with new ImpactFunc.__init__

* white space cosmetics

* tutorials: undo re-plotting

* white space cosmetics

Co-authored-by: Emanuel Schmid <[email protected]>
Co-authored-by: emanuel-schmid <[email protected]>
  • Loading branch information
3 people authored Oct 28, 2022
1 parent c7af81c commit 06a35a1
Show file tree
Hide file tree
Showing 11 changed files with 246 additions and 288 deletions.
18 changes: 10 additions & 8 deletions climada/engine/unsequa/input_var.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,14 +80,16 @@ def litpop_cat(m, n):
Continuous variable function: Impact function for TC
import scipy as sp
def imp_fun_tc(G, v_half, vmin, k, _id=1):
imp_fun = ImpactFunc()
imp_fun.haz_type = 'TC'
imp_fun.id = _id
imp_fun.intensity_unit = 'm/s'
imp_fun.intensity = np.linspace(0, 150, num=100)
imp_fun.mdd = np.repeat(1, len(imp_fun.intensity))
imp_fun.paa = np.array([sigmoid_function(v, G, v_half, vmin, k)
for v in imp_fun.intensity])
intensity = np.linspace(0, 150, num=100)
mdd = np.repeat(1, len(intensity))
paa = np.array([sigmoid_function(v, G, v_half, vmin, k)
for v in intensity])
imp_fun = ImpactFunc(haz_type='TC',
id=_id,
intensity_unit='m/s',
intensity=intensity,
mdd=mdd,
paa=paa)
imp_fun.check()
impf_set = ImpactFuncSet()
impf_set.append(imp_fun)
Expand Down
14 changes: 7 additions & 7 deletions climada/engine/unsequa/test/test_unsequa.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,13 +51,13 @@


def impf_dem(x_paa=1, x_mdd=1):
impf = ImpactFunc()
impf.haz_type = 'TC'
impf.id = 1
impf.intensity_unit = 'm/s'
impf.intensity = np.linspace(0, 150, num=100)
impf.mdd = np.repeat(1, len(impf.intensity)) * x_mdd
impf.paa = np.arange(0, len(impf.intensity)) / len(impf.intensity) * x_paa
haz_type = 'TC'
id = 1
intensity_unit = 'm/s'
intensity = np.linspace(0, 150, num=100)
mdd = np.repeat(1, len(intensity)) * x_mdd
paa = np.arange(0, len(intensity)) / len(intensity) * x_paa
impf = ImpactFunc(haz_type, id, intensity, mdd, paa, intensity_unit)
impf.check()
impf_set = ImpactFuncSet()
impf_set.append(impf)
Expand Down
74 changes: 50 additions & 24 deletions climada/entity/impact_funcs/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,15 @@
__all__ = ['ImpactFunc']

import logging
from typing import Optional, Union
import numpy as np
import matplotlib.pyplot as plt

import climada.util.checker as u_check

LOGGER = logging.getLogger(__name__)


class ImpactFunc():
"""Contains the definition of one impact function.
Expand All @@ -52,16 +54,47 @@ class ImpactFunc():
percentage of affected assets (exposures) for each
intensity (numbers in [0,1])
"""
def __init__(self):
"""Empty initialization."""
self.id = ''
self.name = ''
self.intensity_unit = ''
self.haz_type = ''

def __init__(
self,
haz_type: str = "",
id: Union[str, int] = "",
intensity: Optional[np.ndarray] = None,
mdd: Optional[np.ndarray] = None,
paa: Optional[np.ndarray] = None,
intensity_unit: str = "",
name: str = "",
):
"""Initialization.
Parameters
----------
haz_type : str, optional
Hazard type acronym (e.g. 'TC').
id : int or str, optional
id of the impact function. Exposures of the same type
will refer to the same impact function id.
intensity : np.array, optional
Intensity values. Defaults to empty array.
mdd : np.array, optional
Mean damage (impact) degree for each intensity (numbers
in [0,1]). Defaults to empty array.
paa : np.array, optional
Percentage of affected assets (exposures) for each
intensity (numbers in [0,1]). Defaults to empty array.
intensity_unit : str, optional
Unit of the intensity.
name : str, optional
Name of the ImpactFunc.
"""
self.id = id
self.name = name
self.intensity_unit = intensity_unit
self.haz_type = haz_type
# Followng values defined for each intensity value
self.intensity = np.array([])
self.mdd = np.array([])
self.paa = np.array([])
self.intensity = intensity if intensity is not None else np.array([])
self.mdd = mdd if mdd is not None else np.array([])
self.paa = paa if paa is not None else np.array([])

def calc_mdr(self, inten):
"""Interpolate impact function to a given intensity.
Expand Down Expand Up @@ -156,17 +189,14 @@ def from_step_impf(cls, intensity, mdd=(0, 1), paa=(1, 1), impf_id=1):
Step impact function
"""

impf = cls()
impf.id = impf_id
inten_min, threshold, inten_max = intensity
impf.intensity = np.array([inten_min, threshold, threshold, inten_max])
intensity = np.array([inten_min, threshold, threshold, inten_max])
paa_min, paa_max = paa
impf.paa = np.array([paa_min, paa_min, paa_max, paa_max])
paa = np.array([paa_min, paa_min, paa_max, paa_max])
mdd_min, mdd_max = mdd
impf.mdd = np.array([mdd_min, mdd_min, mdd_max, mdd_max])
mdd = np.array([mdd_min, mdd_min, mdd_max, mdd_max])

return impf
return cls(id=impf_id, intensity=intensity, mdd=mdd, paa=paa)

def set_step_impf(self, *args, **kwargs):
"""This function is deprecated, use ImpactFunc.from_step_impf instead."""
Expand Down Expand Up @@ -208,19 +238,15 @@ def from_sigmoid_impf(cls, intensity, L, k, x0, if_id=1):
Step impact function
"""
impf = cls()
impf.id = if_id
inten_min, inten_max, inten_step = intensity
impf.intensity = np.arange(inten_min, inten_max, inten_step)
impf.paa = np.ones(len(impf.intensity))
impf.mdd = L / (1 + np.exp(-k * (impf.intensity - x0)))
intensity = np.arange(inten_min, inten_max, inten_step)
paa = np.ones(len(intensity))
mdd = L / (1 + np.exp(-k * (intensity - x0)))

return impf
return cls(id=if_id, intensity=intensity, paa=paa, mdd=mdd)

def set_sigmoid_impf(self, *args, **kwargs):
"""This function is deprecated, use LitPop.from_countries instead."""
LOGGER.warning("The use of ImpactFunc.set_sigmoid_impf is deprecated."
"Use ImpactFunc.from_sigmoid_impf instead.")
self.__dict__ = ImpactFunc.from_sigmoid_impf(*args, **kwargs).__dict__


53 changes: 28 additions & 25 deletions climada/entity/impact_funcs/impact_func_set.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,12 +81,12 @@ def __init__(self):
--------
Fill impact functions with values and check consistency data:
>>> fun_1 = ImpactFunc()
>>> fun_1.haz_type = 'TC'
>>> fun_1.id = 3
>>> fun_1.intensity = np.array([0, 20])
>>> fun_1.paa = np.array([0, 1])
>>> fun_1.mdd = np.array([0, 0.5])
>>> fun_1 = ImpactFunc(haz_type, id, intensity, mdd, paa)
>>> imp_fun = ImpactFuncSet()
>>> imp_fun.append(fun_1)
>>> imp_fun.check()
Expand Down Expand Up @@ -435,26 +435,27 @@ def _get_hdf5_str(imp, idxs, file_name, var_name):
imp = imp[var_names['field_name']]
funcs_idx = _get_hdf5_funcs(imp, file_name, var_names)
for imp_key, imp_rows in funcs_idx.items():
func = ImpactFunc()
func.haz_type = imp_key[0]
func.id = imp_key[1]
# Store arguments in a dict (missing ones will be default)
impf_kwargs = dict()
impf_kwargs["haz_type"] = imp_key[0]
impf_kwargs["id"] = imp_key[1]
# check that this function only has one intensity unit, if provided
try:
func.intensity_unit = _get_hdf5_str(imp, imp_rows,
file_name,
var_names['var_name']['unit'])
impf_kwargs["intensity_unit"] = _get_hdf5_str(
imp, imp_rows, file_name, var_names['var_name']['unit'])
except KeyError:
pass
# check that this function only has one name
try:
func.name = _get_hdf5_str(imp, imp_rows, file_name,
var_names['var_name']['name'])
impf_kwargs["name"] = _get_hdf5_str(
imp, imp_rows, file_name, var_names['var_name']['name'])
except KeyError:
func.name = str(func.id)
func.intensity = np.take(imp[var_names['var_name']['inten']], imp_rows)
func.mdd = np.take(imp[var_names['var_name']['mdd']], imp_rows)
func.paa = np.take(imp[var_names['var_name']['paa']], imp_rows)
impf_set.append(func)
impf_kwargs["name"] = str(impf_kwargs["idx"])
impf_kwargs["intensity"] = np.take(
imp[var_names['var_name']['inten']], imp_rows)
impf_kwargs["mdd"] = np.take(imp[var_names['var_name']['mdd']], imp_rows)
impf_kwargs["paa"] = np.take(imp[var_names['var_name']['paa']], imp_rows)
impf_set.append(ImpactFunc(**impf_kwargs))
except KeyError as err:
raise KeyError("Not existing variable: %s" % str(err)) from err

Expand Down Expand Up @@ -523,32 +524,34 @@ def _get_xls_funcs(dfr, var_names):
df_func = df_func[df_func[var_names['col_name']['func_id']]
== imp_id]

func = ImpactFunc()
func.haz_type = haz_type
func.id = imp_id
# Store arguments in a dict (missing ones will be default)
impf_kwargs = dict()
impf_kwargs["haz_type"] = haz_type
impf_kwargs["id"] = imp_id
# check that the unit of the intensity is the same
try:
if len(df_func[var_names['col_name']['name']].unique()) != 1:
raise ValueError('Impact function with two different names.')
func.name = df_func[var_names['col_name']['name']].values[0]
impf_kwargs["name"] = df_func[var_names['col_name']
['name']].values[0]
except KeyError:
func.name = str(func.id)
impf_kwargs["name"] = str(impf_kwargs["id"])

# check that the unit of the intensity is the same, if provided
try:
if len(df_func[var_names['col_name']['unit']].unique()) != 1:
raise ValueError('Impact function with two different \
intensity units.')
func.intensity_unit = \
df_func[var_names['col_name']['unit']].values[0]
impf_kwargs["intensity_unit"] = df_func[var_names['col_name']
['unit']].values[0]
except KeyError:
pass

func.intensity = df_func[var_names['col_name']['inten']].values
func.mdd = df_func[var_names['col_name']['mdd']].values
func.paa = df_func[var_names['col_name']['paa']].values
impf_kwargs["intensity"] = df_func[var_names['col_name']['inten']].values
impf_kwargs["mdd"] = df_func[var_names['col_name']['mdd']].values
impf_kwargs["paa"] = df_func[var_names['col_name']['paa']].values

self.append(func)
self.append(ImpactFunc(**impf_kwargs))

except KeyError as err:
raise KeyError("Not existing variable: %s" % str(err)) from err
8 changes: 4 additions & 4 deletions climada/entity/impact_funcs/test/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,10 @@ class TestInterpolation(unittest.TestCase):

def test_calc_mdr_pass(self):
"""Compute mdr interpolating values."""
imp_fun = ImpactFunc()
imp_fun.intensity = np.arange(0, 100, 10)
imp_fun.paa = np.arange(0, 1, 0.1)
imp_fun.mdd = np.arange(0, 1, 0.1)
intensity = np.arange(0, 100, 10)
paa = np.arange(0, 1, 0.1)
mdd = np.arange(0, 1, 0.1)
imp_fun = ImpactFunc(intensity=intensity, paa=paa, mdd=mdd)
new_inten = 17.2
self.assertEqual(imp_fun.calc_mdr(new_inten), 0.029583999999999996)

Expand Down
Loading

0 comments on commit 06a35a1

Please sign in to comment.