diff --git a/climada/entity/measures/helper.py b/climada/entity/measures/helper.py index af61651b1..771c9deee 100644 --- a/climada/entity/measures/helper.py +++ b/climada/entity/measures/helper.py @@ -23,6 +23,7 @@ import logging from functools import reduce +from pathlib import Path from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar, cast import numpy as np @@ -274,6 +275,7 @@ def helper_hazard(hazard_modifier: HazardModifierConfig) -> HazardChange: """ def hazard_change(hazard: Hazard, **_kwargs) -> Hazard: + changed_hazard = ( Hazard.from_hdf5(hazard_modifier.new_hazard_path) if hazard_modifier.new_hazard_path is not None @@ -304,6 +306,15 @@ def hazard_change(hazard: Hazard, **_kwargs) -> Hazard: hazard_change, ) + if hazard_modifier.new_hazard_path is not None: + if not isinstance(hazard_modifier.new_hazard_path, (str, Path)): + raise ValueError( + f"The new hazard path has incorrect type: {type(hazard_modifier.new_hazard_path)}" + ) + + if hazard_modifier.new_hazard_path == "nil": + hazard_modifier.new_hazard_path = None + return hazard_change @@ -395,6 +406,15 @@ def impfset_change(impfset: ImpactFuncSet, **_kwargs) -> ImpactFuncSet: return changed_impfset + if impfset_modifier.new_impfset_path is not None: + if not isinstance(impfset_modifier.new_impfset_path, (str, Path)): + raise ValueError( + f"The new impfset path has incorrect type: {type(impfset_modifier.new_impfset_path)}" + ) + + if impfset_modifier.new_impfset_path == "nil": + impfset_modifier.new_impfset_path = None + return impfset_change @@ -473,4 +493,13 @@ def exposures_change(exposures: Exposures, **_kwargs) -> Exposures: return changed_exposures + if exposures_modifier.new_exposures_path is not None: + if not isinstance(exposures_modifier.new_exposures_path, (str, Path)): + raise ValueError( + f"The new exposures path has incorrect type: {type(exposures_modifier.new_exposures_path)}" + ) + + if exposures_modifier.new_exposures_path == "nil": + exposures_modifier.new_exposures_path = None + return exposures_change diff --git a/climada/entity/measures/measure.py b/climada/entity/measures/measure.py index 38db9f7fc..1c095723a 100755 --- a/climada/entity/measures/measure.py +++ b/climada/entity/measures/measure.py @@ -265,6 +265,34 @@ def from_dict(cls, d: dict) -> "Measure": """ return cls.from_config(MeasureConfig.from_dict(d)) + def to_dict(self) -> dict: + """Serialize the measure to a dictionary. + + Only measures created from a ``MeasureConfig`` (see + :attr:`is_serializable`) can be serialized. The result is the flat + dictionary returned by :py:meth:`MeasureConfig.to_dict`, so it can be + read back with :py:meth:`Measure.from_dict`. + + Returns + ------- + dict + Flat dictionary representation of the measure configuration. + + Raises + ------ + ValueError + If the measure was not created from a ``MeasureConfig``, for + instance when it was defined with Python functions. + """ + if self._config is None: + raise ValueError( + f"Measure '{self.name}' cannot be serialized: it was not created " + "from a MeasureConfig (Python change functions cannot be " + "serialized). Define it with Measure.from_config, from_dict " + "or a MeasureConfig instead." + ) + return self._config.to_dict() + def apply_exposures_changes( self, exposures: Exposures, enforce_copy: bool = True, **kwargs ) -> Exposures: diff --git a/climada/entity/measures/measure_set.py b/climada/entity/measures/measure_set.py new file mode 100644 index 000000000..b6de37ae5 --- /dev/null +++ b/climada/entity/measures/measure_set.py @@ -0,0 +1,651 @@ +""" +This file is part of CLIMADA. + +Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS. + +CLIMADA is free software: you can redistribute it and/or modify it under the +terms of the GNU General Public License as published by the Free +Software Foundation, version 3. + +CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with CLIMADA. If not, see . + +--- + +Define MeasureSet class. +""" + +import copy +from typing import no_type_check + +__all__ = ["MeasureSet"] + +import json +import logging +from collections.abc import Iterable +from functools import reduce + +import geopandas as gpd +import numpy as np +import pandas as pd +import yaml + +import climada.util.hdf5_handler as u_hdf5 +from climada.entity.exposures.base import Exposures +from climada.entity.impact_funcs import ImpactFunc, ImpactFuncSet +from climada.entity.measures.cost_income import CostIncome +from climada.entity.measures.helper import composite_fun +from climada.entity.measures.measure import Measure +from climada.entity.measures.measure_config import MeasureConfig +from climada.hazard.base import Hazard +from climada.util.string_parsers import parse_mapping_string + +LOGGER = logging.getLogger(__name__) + +DEF_VAR_MAT = { + "sup_field_name": "entity", + "field_name": "measures", + "var_name": { + "name": "name", + "color": "color", + "cost": "cost", + "haz_int_a": "hazard_intensity_impact_a", + "haz_int_b": "hazard_intensity_impact_b", + "haz_frq": "hazard_high_frequency_cutoff", + "haz_set": "hazard_event_set", + "mdd_a": "MDD_impact_a", + "mdd_b": "MDD_impact_b", + "paa_a": "PAA_impact_a", + "paa_b": "PAA_impact_b", + "fun_map": "damagefunctions_map", + "exp_set": "assets_file", + "exp_reg": "Region_ID", + "risk_att": "risk_transfer_attachement", + "risk_cov": "risk_transfer_cover", + "haz": "peril_ID", + }, +} +"""MATLAB variable names""" + +DEF_VAR_EXCEL = { + "sheet_name": "measures", + "col_name": { + "name": "name", + "color": "color_rgb", + "implementation duration": "implementation_duration", + "cost": "init_cost", + "periodic cost": "periodic_cost", + "periodic income": "periodic_income", + "income growth rate (yearly)": "income_yearly_growth_rate", + "cost growth rate (yearly)": "cost_yearly_growth_rate", + "impact function id": "impf_id", + "hazard intensity impact a": "haz_int_mult", + "hazard intensity impact b": "haz_int_add", + "hazard event set": "haz_set", + "MDD impact a": "impf_mdd_mult", + "MDD impact b": "impf_mdd_add", + "PAA impact a": "impf_paa_mult", + "PAA impact b": "imfp_paa_add", + "damagefunctions map": "fun_map", + "assets file": "exp_set", + "Region_ID": "exp_reg", + "peril_ID": "haz_type", + "Impact RP cutoff": "impact_rp_cutoff", + "assets zeroing": "assets_to_zero", + }, +} +"""Excel variable names""" + + +class MeasureSet: + """Contains measures of type Measure. + + Attributes + ---------- + _data : dict + Contains Measure objects keyed by their name. + """ + + def __init__(self, measures: Iterable[Measure]): + """Initialize a new MeasureSet object. + + Parameters + ---------- + measures : Iterable[Measure]. + The measures to include in the MeasureSet. + + """ + self._data: dict[str, Measure] = {meas.name: meas for meas in measures} + + def append(self, measure: Measure): + """ + Append a Measure. Overwrites if a measure with the same name exists. + + Parameters + ---------- + measure : Measure + The Measure instance to add. + + Raises + ------ + TypeError + If input is not an instance of Measure. + + """ + if not isinstance(measure, Measure): + raise TypeError(f"Expected Measure, got {type(measure).__name__}") + + self._data[measure.name] = measure + + def select(self, names: str | Iterable[str]) -> dict[str, Measure]: + """ + Get a dictionary of measures. + + Parameters + ---------- + names : str|[str] + Filter by these measure names. + + Returns + ------- + dict[str, Measure] + dictionary of measure names and objects. + + """ + if isinstance(names, str): + name_list = [names] + else: + name_list = list(names) + + missing = [name for name in name_list if name not in self._data] + if missing: + raise ValueError(f"No measure named {missing} in the measure set.") + + return {name: self._data[name] for name in name_list} + + @property + def names(self) -> list[str]: + """Return the list of the measure names""" + return list(self._data.keys()) + + def __len__(self) -> int: + """Number of measures in the set.""" + return len(self._data) + + @property + def size(self) -> int: + """Number of measures in the set.""" + return self.__len__() + + def __contains__(self, item: str) -> bool: + """Check if a measure name exists in the set.""" + return item in self._data + + def compose(self, names: list[str], combo_name: str | None = None) -> Measure: + """ + Compose multiple measures into a single meta-Measure. + + This method creates a new Measure where the transformation functions + (hazard, exposures, impfset) are mathematically composed and financial + values are aggregated. + + Execution Order: + The composition follows a right-to-left nesting. For a list of + measures [m1, m2, m3], the resulting transformation is: + m1(m2(m3(x))) + This means m3 is applied first, then m2, then m1. + + Financial Aggregation: + - Costs and incomes are summed across all selected measures. + - Growth rates and market price years are inherited from the + first measure in the list. + + Parameters + ---------- + names : list of str + Ordered list of measure names to combine. + combo_name : str, optional + The name for the resulting Measure. If None, a name is + generated by joining names with an underscore and + appending 'composed'. + + Returns + ------- + Measure + A new Measure object representing the sequential application + of the component measures. + + Raises + ------ + ValueError + If the provided names do not match any measures in the set, + or if the MeasureSet is empty. + + """ + meas_list = list(self.select(names).values()) + + if not meas_list: + raise ValueError("No measures found to compose.") + + def exposure_changes(exp: Exposures, **kwargs) -> Exposures: + return composite_fun(*[meas.exposures_changes for meas in meas_list])( + exp, **kwargs + ) + + def impfset_changes(impfset: ImpactFuncSet, **kwargs) -> ImpactFuncSet: + return composite_fun(*[meas.impfset_changes for meas in meas_list])( + impfset, **kwargs + ) + + def hazard_changes(haz: Hazard, **kwargs) -> Hazard: + return composite_fun(*[meas.hazard_changes for meas in meas_list])( + haz, **kwargs + ) + + return Measure( + name=combo_name or "_".join(names) + "_composed", + exposures_changes=exposure_changes, + impfset_changes=impfset_changes, + hazard_changes=hazard_changes, + sub_measures=names, + cost_income=CostIncome.comb_cost_income([m.cost_income for m in meas_list]), + ) + + @staticmethod + def _combine_hazards(modified_hazards: list[Hazard]) -> Hazard: + """Finds the maximum effect (minimum intensity/freq) across hazards.""" + intensities = [h.intensity for h in modified_hazards] + fractions = [h.fraction for h in modified_hazards] + frequencies = [h.frequency for h in modified_hazards] + + hazard_mod = copy.deepcopy(modified_hazards[0]) + hazard_mod.intensity = reduce(lambda a, b: a.minimum(b), intensities) + hazard_mod.fraction = reduce(lambda a, b: a.minimum(b), fractions) + hazard_mod.frequency = np.minimum.reduce(frequencies) + return hazard_mod + + @staticmethod + def _combine_impfsets( + base_set: ImpactFuncSet, modified_sets: list[ImpactFuncSet] + ) -> ImpactFuncSet: + """Merges impact functions by taking the safest (minimum) damage parameters.""" + combined = ImpactFuncSet() + for haz_dict in base_set.get_func().values(): + for impf in haz_dict.values(): + versions = [ + s.get_func(haz_type=impf.haz_type, fun_id=impf.id) + for s in modified_sets + ] + + combined.append( + ImpactFunc( + impf.haz_type, + impf.id, + intensity=np.maximum.reduce([v.intensity for v in versions]), + mdd=np.minimum.reduce([v.mdd for v in versions]), + paa=np.minimum.reduce([v.paa for v in versions]), + intensity_unit=impf.intensity_unit, + name=impf.name, + ) + ) + return combined + + @staticmethod + def _combine_exposures( + base_exp: Exposures, modified_exps: list[Exposures] + ) -> Exposures: + """Merges exposure changes, raising ValueError if two measures touch the same cell.""" + new_exps_gdfs = [exp.gdf for exp in modified_exps] + if not all( + set(new_gdf.columns) == set(base_exp.gdf.columns) + for new_gdf in new_exps_gdfs + ): + raise ValueError( + "All change DataFrames must have identical column structure and order." + ) + + if not all( + new_gdf.index.equals(base_exp.gdf.index) for new_gdf in new_exps_gdfs + ): + raise ValueError("All change DataFrames must have the same index.") + + # Align all changes into a single MultiIndexed DataFrame + # This stacks all change-sets on top of each other + stack = pd.concat( + new_exps_gdfs, + keys=range(len(new_exps_gdfs)), + names=["change_idx", "row_idx"], + ) + + # Create a broadcasted baseline to match the stack's shape + # We use take() to repeat baseline rows for every change-set + baseline_repeated = base_exp.gdf.iloc[ + np.tile(np.arange(len(base_exp.gdf)), len(modified_exps)) + ] + baseline_repeated.index = stack.index # Align indices for direct comparison + + # Identify changes: Mask is True where a cell differs from baseline + # Note that a != b is true if a.isna() and b.isna(), so it is trickier + diff_mask = ~( + (stack == baseline_repeated) | (stack.isna() & baseline_repeated.isna()) + ) + + # Check for Conflicts: + # Sum the True values across the 'change_idx' level for every (row, col) + # If any cell has > 1 change, it's a conflict. + change_counts = diff_mask.groupby(level="row_idx").sum() + if (change_counts > 1).any().any(): + # Identify exactly where the conflict is for the error message + conflicting_cells = ( + change_counts[change_counts > 1] + .dropna(how="all") + .dropna(axis=1, how="all") + ) + raise ValueError( + f"Conflict: Multiple measures change the same cells:\n{conflicting_cells}" + ) + + # Merge: + # We take the baseline and update it with the sum of differences + # Only works if the data is numeric. For general objects (like 'if_' IDs): + result = base_exp.gdf.copy() + geometry = result["geometry"] + + # Efficiently collapse the stack: + # Since only one change exists per cell (checked in step 5), + # we can 'first' or 'max' to get the non-baseline value. + updates = stack.where(diff_mask).groupby(level="row_idx").first() + updates = updates.combine_first(result) + + # combine_first is pure DataFrame + new_gdf = gpd.GeoDataFrame(updates, geometry=geometry, crs=result.crs) + + exp_mod = Exposures( + new_gdf, + crs=base_exp.crs, + description=base_exp.description, + ref_year=base_exp.ref_year, + value_unit=base_exp.value_unit, + ) + return exp_mod + + def combine( + self, names: list[str] | None = None, combo_name: str | None = None + ) -> Measure: + """Combine multiple measures into a single composite Measure object. + + This method creates a new Measure that applies all specified sub-measures + sequentially to exposures, impact function sets, and hazard data. Each + sub-measure's transformation is applied individually, then the results are + aggregated through specialized combination methods that implement "best-of-all" + principles (taking minimum damage parameters, maximum effects). + + The combined Measure preserves the cost/income information from all sub-measures. + + Parameters + ---------- + names : list[str], optional + List of measure names to subset before combining. If None, uses all + measures. Each name must correspond to an existing measure in the + collection. + combo_name : str, optional + Custom name for the combined measure. If None, defaults to joining the + sub-measure names with underscores (e.g., "measure1_measure2_measure3"). + + Returns + ------- + Measure + A new Measure object containing: + + * Combined transformation functions for exposures, impact function sets, + and hazard data + * Aggregated cost/income information from all sub-measures + * Reference to all sub-measure names for traceability + + Raises + ------ + ValueError + If no measures are found to combine (empty `names` list or no valid + measures). Also raised if: + + * Multiple measures attempt to modify the same exposure cells (conflict + detected during exposure combination) + * Cost/income objects have mismatched market price years, cost growth + rates, or income growth rates + + Notes + ----- + Combination Logic by Entity Type: + + **Hazards**: Maximum effect is taken across all modified hazards, implemented + as minimum intensity, minimum fraction, and minimum frequency values. This + represents the most conservative (highest impact) scenario. + + **Impact Functions**: Merged by taking the safest (minimum) damage parameters + (MDD and PAA) across all modified impact functions, while preserving the + maximum intensity range. This ensures conservative damage estimation. + + **Exposures**: Changes are merged with strict conflict detection. If multiple + measures attempt to modify the same exposure cell, a ValueError is raised. + All change DataFrames must have identical column structure and order. + + **Cost/Income**: Values are summed across all sub-measures. Validation ensures + all cost/income objects share the same market price year, cost growth rate, + and income growth rate before aggregation. + + The wrapper functions preserve `**kwargs` support, allowing entity-specific + configuration to be passed through to individual sub-measure transformations. + This enables fine-grained control over how each sub-measure behaves during + the combined operation. + """ + names = self.names if names is None else names + meas_list = list(self.select(names).values()) + + if not meas_list: + raise ValueError("No measures found to combine.") + + def combined_exposure_changes(base_exp: Exposures, **kwargs) -> Exposures: + # 1. Apply all measures individually + mod_exps = [ + m.apply_exposures_changes(base_exp, **kwargs) for m in meas_list + ] + + # 2. Delegate combination to specialized methods + return self._combine_exposures(base_exp, mod_exps) + + def combined_impfset_changes(base_impfs: ImpactFuncSet, **kwargs): + # 1. Apply all measures individually + mod_impfs = [ + m.apply_impfset_changes(base_impfs, **kwargs) for m in meas_list + ] + + # 2. Delegate combination to specialized methods + return self._combine_impfsets(base_impfs, mod_impfs) + + def combined_hazard_changes(base_haz: Hazard, **kwargs): + # 1. Apply all measures individually + mod_haz = [m.apply_hazard_changes(base_haz, **kwargs) for m in meas_list] + + # 2. Delegate combination to specialized methods + return self._combine_hazards(mod_haz) + + return Measure( + name=combo_name or "_".join(names), + sub_measures=names, + exposures_changes=combined_exposure_changes, + impfset_changes=combined_impfset_changes, + hazard_changes=combined_hazard_changes, + cost_income=CostIncome.comb_cost_income([m.cost_income for m in meas_list]), + ) + + @no_type_check + @classmethod + def from_mat(cls, file_name: str, var_names: dict | None = None) -> "MeasureSet": + """Read MATLAB file generated with previous MATLAB CLIMADA version. + + Parameters + ---------- + file_name : str + Absolute path to the MATLAB file. + var_names : dict, optional + Name of the variables in the file. Defaults to DEF_VAR_MAT. + + Returns + ------- + MeasureSet + """ + if var_names is None: + var_names = DEF_VAR_MAT + + def _parse_measure(idx: int, data: dict) -> MeasureConfig: + vn = var_names["var_name"] + + haz_type = u_hdf5.get_str_from_ref(file_name, data[vn["haz"]][idx][0]) + impf_id = 1 # MATLAB format has no explicit impf_id + + # hazard intensity: old files may lack the _a/_b suffix + try: + haz_int_a = data[vn["haz_int_a"]][idx][0] + haz_int_b = data[vn["haz_int_b"]][0][idx] + except KeyError: + haz_int_a = data[vn["haz_int_a"][:-2]][idx][0] + haz_int_b = 0.0 + + # optional fields that may be empty strings in legacy files + haz_set = ( + u_hdf5.get_str_from_ref(file_name, data[vn["haz_set"]][idx][0]) or None + ) + exp_set = ( + u_hdf5.get_str_from_ref(file_name, data[vn["exp_set"]][idx][0]) or None + ) + fun_map = ( + u_hdf5.get_str_from_ref(file_name, data[vn["fun_map"]][idx][0]) or None + ) + color_str = u_hdf5.get_str_from_ref(file_name, data[vn["color"]][idx][0]) + + if data[vn["exp_reg"]][idx][0]: + LOGGER.warning( + "Measure '%s' has exp_region_id set, which is no longer supported " + "and will be ignored. It will be reimplemented in a future version.", + u_hdf5.get_str_from_ref(file_name, data[vn["name"]][idx][0]), + ) + + if data[vn["risk_att"]][idx][0] or data[vn["risk_cov"]][idx][0]: + LOGGER.warning( + "Measure '%s' has risk_transf_attach/cover set, which is no longer " + "supported and will be ignored. It will be reimplemented in a future version.", + u_hdf5.get_str_from_ref(file_name, data[vn["name"]][idx][0]), + ) + + return MeasureConfig.from_dict( + { + "name": u_hdf5.get_str_from_ref( + file_name, data[vn["name"]][idx][0] + ), + "haz_type": haz_type, + "impf_id": impf_id, + "impf_mdd_mult": float(data[vn["mdd_a"]][idx][0]), + "impf_mdd_add": float(data[vn["mdd_b"]][idx][0]), + "impf_paa_mult": float(data[vn["paa_a"]][idx][0]), + "impf_paa_add": float(data[vn["paa_b"]][idx][0]), + "intensity_multiplier": float(haz_int_a), + "intensity_add": float(haz_int_b), + "new_hazard_path": haz_set if haz_set != "nil" else None, + "impact_rp_cutoff": float(data[vn["haz_frq"]][idx][0]) or None, + "reassign_impf_id": ( + parse_mapping_string(fun_map) + if fun_map and fun_map != "nil" + else None + ), + "new_exposures_path": exp_set if exp_set != "nil" else None, + "init_cost": float(data[vn["cost"]][idx][0]), + "color_rgb": tuple(np.fromstring(color_str, dtype=float, sep=" ")), + } + ) + + data = u_hdf5.read(file_name) + try: + data = data[var_names["sup_field_name"]] + except KeyError: + pass + try: + data = data[var_names["field_name"]] + except KeyError as err: + raise KeyError("Variable not in MAT file: " + str(err)) from err + + num_measures = len(data[var_names["var_name"]["name"]]) + measures = [ + Measure.from_config(_parse_measure(idx, data)) + for idx in range(num_measures) + ] + return cls(measures) + + @classmethod + def from_excel(cls, file_name: str, var_names: dict | None = None): + """Read excel file following template and return a MeasureSet.""" + if var_names is None: + var_names = DEF_VAR_EXCEL + + df = pd.read_excel(file_name, sheet_name=var_names["sheet_name"]) # ty: ignore + # inv_map = {v: k for k, v in var_names["col_name"].items()} + df = df.rename(columns=var_names["col_name"]) + + # Extract row processing to reduce locals in main method + measures = [cls._process_excel_row(row) for _, row in df.iterrows()] + return cls(measures) + + @classmethod + def _process_excel_row(cls, row: pd.Series) -> Measure: + """Process a single Excel row into a Measure object.""" + return Measure.from_config(MeasureConfig.from_row(row)) + + def to_dict(self) -> dict: + """Serialize all serializable measures to a dict. Skips function-only measures.""" + serializable = { + name: measure.to_dict() + for name, measure in self._data.items() + if measure.is_serializable + } + skipped = [ + name for name, measure in self._data.items() if not measure.is_serializable + ] + if skipped: + LOGGER.warning( + "The following measures are not serializable and will be skipped: %s", + skipped, + ) + return {"measures": list(serializable.values())} + + @classmethod + def from_dict(cls, d: dict) -> "MeasureSet": + """Build a MeasureSet from a dictionary configuration.""" + measures = [ + Measure.from_config(MeasureConfig.from_dict(m)) for m in d["measures"] + ] + return cls(measures) + + def to_yaml(self, path: str) -> None: + """Write the MeasureSet to a yaml file. Only serializable measures are written.""" + with open(path, "w", encoding="utf-8") as f: + yaml.dump(self.to_dict(), f, default_flow_style=False, sort_keys=False) + + @classmethod + def from_yaml(cls, path: str) -> "MeasureSet": + """Read a MeasureSet from a yaml file.""" + with open(path, "r", encoding="utf-8") as f: + return cls.from_dict(yaml.safe_load(f)) + + def to_json(self, path: str) -> None: + """Write the MeasureSet to a json file. Only serializable measures are written.""" + with open(path, "w", encoding="utf-8") as f: + json.dump(self.to_dict(), f, indent=2) + + @classmethod + def from_json(cls, path: str) -> "MeasureSet": + """Read a MeasureSet from a json file.""" + with open(path, "r", encoding="utf-8") as f: + return cls.from_dict(json.load(f)) diff --git a/climada/entity/measures/test/test_measure_set.py b/climada/entity/measures/test/test_measure_set.py new file mode 100644 index 000000000..245b2c0ef --- /dev/null +++ b/climada/entity/measures/test/test_measure_set.py @@ -0,0 +1,471 @@ +""" +This file is part of CLIMADA. + +Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS. + +CLIMADA is free software: you can redistribute it and/or modify it under the +terms of the GNU General Public License as published by the Free +Software Foundation, version 3. + +CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with CLIMADA. If not, see . + +--- + +Unit tests for the MeasureSet class. +""" + +import json +from unittest.mock import MagicMock, patch + +import numpy as np +import pandas as pd +import pytest +import yaml + +from climada.entity.measures.cost_income import CostIncome +from climada.entity.measures.measure import Measure +from climada.entity.measures.measure_config import MeasureConfig +from climada.entity.measures.measure_set import MeasureSet + +MODULE = "climada.entity.measures.measure_set" +NAMES = ["flood_barrier", "sea_wall", "early_warning"] + + +def _make_mock(): + """Return a MagicMock standing in for Exposures / ImpactFuncSet / Hazard.""" + return MagicMock() + + +def _make_measure(name, init_cost=0.0, **kwargs): + """Return a lightweight Measure. + + Parameters + ---------- + name : str + Measure name. + init_cost : float, optional + Initial cost of the measure. Default is 0. + **kwargs + Forwarded to `Measure`. + + Returns + ------- + Measure + """ + return Measure(name, cost_income=CostIncome(init_cost=init_cost), **kwargs) + + +def _tag(label): + """Return a change function appending `label` to a list-like object. + + Parameters + ---------- + label : str + Label appended when the function is called. + + Returns + ------- + callable + Function ``fn(obj, **kwargs)`` that appends `label` to `obj` and returns it. + """ + + def fn(obj, **kwargs): + obj.append(label) + return obj + + return fn + + +def _single_impf(impfset): + """Return the only ImpactFunc of an ImpactFuncSet.""" + haz_dict = next(iter(impfset.get_func().values())) + return next(iter(haz_dict.values())) + + +@pytest.fixture +def measure_set(): + return MeasureSet( + [_make_measure(name, init_cost=100.0 * (i + 1)) for i, name in enumerate(NAMES)] + ) + + +class TestInit: + def test_names_and_size(self, measure_set): + assert measure_set.names == NAMES + assert measure_set.size == 3 + + def test_empty(self): + ms = MeasureSet([]) + assert ms.names == [] + assert ms.size == 0 + + def test_duplicate_names_keep_last(self): + first, last = _make_measure("dup"), _make_measure("dup") + ms = MeasureSet([first, last]) + assert ms.size == 1 + assert ms.select(["dup"])["dup"] is last + + def test_accepts_generator(self): + ms = MeasureSet(_make_measure(n) for n in NAMES) + assert ms.names == NAMES + + +class TestAppend: + def test_adds_measure(self, measure_set): + measure_set.append(_make_measure("new")) + assert "new" in measure_set + assert measure_set.size == 4 + + def test_overwrites_same_name(self, measure_set): + new = _make_measure(NAMES[0]) + measure_set.append(new) + assert measure_set.size == 3 + assert measure_set.select([NAMES[0]])[NAMES[0]] is new + + def test_non_measure_raises_type_error(self, measure_set): + with pytest.raises(TypeError, match="Expected Measure, got str"): + measure_set.append("not a measure") + + +class TestSelect: + def test_returns_requested_measures_in_order(self, measure_set): + result = measure_set.select([NAMES[2], NAMES[0]]) + assert list(result) == [NAMES[2], NAMES[0]] + assert all(isinstance(m, Measure) for m in result.values()) + + def test_unknown_name_raises_value_error(self, measure_set): + with pytest.raises(ValueError, match="No measure named"): + measure_set.select(["unknown"]) + + def test_empty_selection(self, measure_set): + assert measure_set.select([]) == {} + + +class TestContains: + def test_present(self, measure_set): + assert NAMES[0] in measure_set + + def test_absent(self, measure_set): + assert "unknown" not in measure_set + + +class TestCompose: + def test_default_name_and_sub_measures(self, measure_set): + result = measure_set.compose(NAMES[:2]) + assert result.name == f"{NAMES[0]}_{NAMES[1]}_composed" + assert result.sub_measures == NAMES[:2] + + def test_custom_name(self, measure_set): + assert measure_set.compose(NAMES[:2], combo_name="combo").name == "combo" + + def test_costs_are_summed(self, measure_set): + result = measure_set.compose(NAMES[:2]) + assert result.cost_income.init_cost == -300.0 + + def test_empty_names_raises_value_error(self, measure_set): + with pytest.raises(ValueError, match="No measures found to compose"): + measure_set.compose([]) + + def test_unknown_name_raises_value_error(self, measure_set): + with pytest.raises(ValueError, match="No measure named"): + measure_set.compose(["unknown"]) + + @pytest.mark.parametrize( + "attr", ["exposures_changes", "impfset_changes", "hazard_changes"] + ) + def test_execution_order_is_right_to_left(self, attr): + ms = MeasureSet( + [_make_measure(f"m{i}", **{attr: _tag(f"m{i}")}) for i in (1, 2, 3)] + ) + composed = ms.compose(["m1", "m2", "m3"]) + assert getattr(composed, attr)([]) == ["m3", "m2", "m1"] + + +class TestCombineHazards: + def test_takes_minimum_of_intensity_and_frequency(self, hazard_factory): + ref = hazard_factory() + low = hazard_factory(intensity_scale=0.5, frequency_scale=2.0) + high = hazard_factory(intensity_scale=1.0, frequency_scale=0.5) + + result = MeasureSet._combine_hazards([low, high]) + + np.testing.assert_allclose( + result.intensity.toarray(), 0.5 * ref.intensity.toarray() + ) + np.testing.assert_allclose(result.frequency, 0.5 * ref.frequency) + assert result.fraction.shape == ref.fraction.shape + + def test_single_hazard_unchanged(self, hazard_factory): + ref = hazard_factory() + result = MeasureSet._combine_hazards([hazard_factory()]) + np.testing.assert_allclose(result.intensity.toarray(), ref.intensity.toarray()) + np.testing.assert_allclose(result.frequency, ref.frequency) + + +class TestCombineImpfsets: + def test_safest_parameters_and_widest_intensity(self, impfset_factory): + base = impfset_factory() + mod_a = impfset_factory() + mod_b = impfset_factory(max_intensity=200, paa_scale=0.5) + _single_impf(mod_b).mdd = np.array([0, 0.25, 0.5]) + + result = MeasureSet._combine_impfsets(base, [mod_a, mod_b]) + + base_impf = _single_impf(base) + impf = result.get_func(haz_type=base_impf.haz_type, fun_id=base_impf.id) + np.testing.assert_allclose(impf.intensity, [0, 100, 200]) + np.testing.assert_allclose(impf.mdd, [0, 0.25, 0.5]) + np.testing.assert_allclose(impf.paa, [0.5, 0.5, 0.5]) + + def test_metadata_taken_from_base(self, impfset_factory): + base = impfset_factory() + result = MeasureSet._combine_impfsets(base, [impfset_factory()]) + base_impf, impf = _single_impf(base), _single_impf(result) + assert impf.intensity_unit == base_impf.intensity_unit + assert impf.name == base_impf.name + assert impf.id == base_impf.id + + +class TestCombineExposures: + def test_merges_changes_on_distinct_columns(self, exposures_factory): + categories = [1, 1, 2, 1, 1, 3] + base = exposures_factory(categories=categories) + mod_value = exposures_factory(categories=categories, value_factor=2.0) + new_categories = list(categories) + new_categories[2] = 9 + mod_category = exposures_factory(categories=new_categories) + + result = MeasureSet._combine_exposures(base, [mod_value, mod_category]) + + np.testing.assert_allclose(result.gdf["value"], 2.0 * base.gdf["value"]) + np.testing.assert_allclose(result.gdf["category"], new_categories) + assert result.ref_year == base.ref_year + assert result.value_unit == base.value_unit + assert result.description == base.description + + def test_no_change_returns_baseline_values(self, exposures_factory): + base = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + same = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + result = MeasureSet._combine_exposures(base, [same]) + np.testing.assert_allclose(result.gdf["value"], base.gdf["value"]) + + def test_conflicting_changes_raise_value_error(self, exposures_factory): + base = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + mod_a = exposures_factory(categories=[1, 1, 2, 1, 1, 3], value_factor=2.0) + mod_b = exposures_factory(categories=[1, 1, 2, 1, 1, 3], value_factor=3.0) + with pytest.raises(ValueError, match="Conflict"): + MeasureSet._combine_exposures(base, [mod_a, mod_b]) + + def test_different_columns_raise_value_error(self, exposures_factory): + base = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + mod = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + mod.gdf["extra"] = 1 + with pytest.raises(ValueError, match="identical column structure"): + MeasureSet._combine_exposures(base, [mod]) + + def test_different_index_raise_value_error(self, exposures_factory): + base = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + mod = exposures_factory(categories=[1, 1, 2, 1, 1, 3]) + mod.gdf.index = range(10, 16) + with pytest.raises(ValueError, match="same index"): + MeasureSet._combine_exposures(base, [mod]) + + +class TestCombine: + def test_default_name_and_sub_measures(self, measure_set): + result = measure_set.combine() + assert result.name == "_".join(NAMES) + assert result.sub_measures == NAMES + + def test_names_subset_and_custom_name(self, measure_set): + result = measure_set.combine(NAMES[:2], combo_name="combo") + assert result.name == "combo" + assert result.sub_measures == NAMES[:2] + + def test_costs_are_summed(self, measure_set): + assert measure_set.combine().cost_income.init_cost == -600.0 + + def test_empty_names_raises_value_error(self, measure_set): + with pytest.raises(ValueError, match="No measures found to combine"): + measure_set.combine([]) + + def test_unknown_name_raises_value_error(self, measure_set): + with pytest.raises(ValueError, match="No measure named"): + measure_set.combine(["unknown"]) + + @pytest.mark.parametrize( + "method, helper, has_base", + [ + ("apply_exposures_changes", "_combine_exposures", True), + ("apply_impfset_changes", "_combine_impfsets", True), + ("apply_hazard_changes", "_combine_hazards", False), + ], + ) + def test_delegates_to_combination_helper( + self, measure_set, method, helper, has_base + ): + obj = _make_mock() + combo = measure_set.combine() + + with patch.object(MeasureSet, helper) as mock_helper: + result = getattr(combo, method)(obj, enforce_copy=False) + + # identity sub-measures return the input, one modified object per measure + modified = [obj] * len(NAMES) + expected_args = (obj, modified) if has_base else (modified,) + mock_helper.assert_called_once_with(*expected_args) + assert result is mock_helper.return_value + + def test_kwargs_forwarded_to_sub_measures(self): + received = {} + + def track(obj, **kwargs): + received.update(kwargs) + return obj + + ms = MeasureSet([_make_measure("a", hazard_changes=track)]) + with patch.object(MeasureSet, "_combine_hazards"): + ms.combine().apply_hazard_changes( + _make_mock(), enforce_copy=False, threshold=3.5 + ) + + assert received.get("threshold") == 3.5 + + +class TestFromExcel: + @pytest.fixture + def var_names(self): + return {"sheet_name": "my_sheet", "col_name": {"cost": "init_cost"}} + + def test_columns_renamed_and_one_measure_per_row(self, var_names): + df = pd.DataFrame({"name": ["a", "b"], "cost": [1.0, 2.0]}) + rows = [] + + def fake_process(row): + rows.append(row) + return Measure(row["name"]) + + with ( + patch(f"{MODULE}.pd.read_excel", return_value=df) as mock_read, + patch.object(MeasureSet, "_process_excel_row", side_effect=fake_process), + ): + ms = MeasureSet.from_excel("file.xlsx", var_names=var_names) + + mock_read.assert_called_once() + assert mock_read.call_args.args == ("file.xlsx",) + assert mock_read.call_args.kwargs["sheet_name"] == "my_sheet" + assert ms.names == ["a", "b"] + assert [r["init_cost"] for r in rows] == [1.0, 2.0] + + def test_default_var_names_used(self): + df = pd.DataFrame({"name": ["a"]}) + with ( + patch(f"{MODULE}.pd.read_excel", return_value=df) as mock_read, + patch.object( + MeasureSet, "_process_excel_row", side_effect=lambda r: Measure("a") + ), + ): + MeasureSet.from_excel("file.xlsx") + + assert mock_read.call_args.kwargs["sheet_name"] == "measures" + + def test_process_excel_row_builds_measure_from_config(self): + row = pd.Series({"name": "a"}) + with ( + patch(f"{MODULE}.MeasureConfig") as mock_config, + patch(f"{MODULE}.Measure") as mock_measure, + ): + result = MeasureSet._process_excel_row(row) + + mock_config.from_row.assert_called_once_with(row) + mock_measure.from_config.assert_called_once_with( + mock_config.from_row.return_value + ) + assert result is mock_measure.from_config.return_value + + +class TestToDict: + def test_serializable_measures_are_dumped(self): + ms = MeasureSet([_make_measure(n, _config=MagicMock()) for n in NAMES[:2]]) + with patch.object( + Measure, "to_dict", autospec=True, side_effect=lambda m: {"name": m.name} + ): + result = ms.to_dict() + + assert result == {"measures": [{"name": n} for n in NAMES[:2]]} + + def test_non_serializable_measures_skipped_with_warning(self): + ms = MeasureSet( + [_make_measure("ok", _config=MagicMock()), _make_measure("function_only")] + ) + with ( + patch.object( + Measure, + "to_dict", + autospec=True, + side_effect=lambda m: {"name": m.name}, + ), + patch(f"{MODULE}.LOGGER") as mock_logger, + ): + result = ms.to_dict() + + assert result == {"measures": [{"name": "ok"}]} + mock_logger.warning.assert_called_once() + assert mock_logger.warning.call_args.args[1] == ["function_only"] + + def test_no_warning_when_all_serializable(self): + ms = MeasureSet([_make_measure("ok", _config=MagicMock())]) + with ( + patch.object(Measure, "to_dict", return_value={}), + patch(f"{MODULE}.LOGGER") as mock_logger, + ): + ms.to_dict() + + mock_logger.warning.assert_not_called() + + +class TestFromDict: + def test_builds_one_measure_per_entry(self): + data = {"measures": [{"name": "a"}, {"name": "b"}]} + with ( + patch.object(MeasureConfig, "from_dict", side_effect=lambda d: d), + patch.object( + Measure, "from_config", side_effect=lambda cfg: Measure(cfg["name"]) + ), + ): + ms = MeasureSet.from_dict(data) + + assert ms.names == ["a", "b"] + + def test_empty_measures(self): + assert MeasureSet.from_dict({"measures": []}).size == 0 + + +class TestFileIO: + PAYLOAD = {"measures": [{"name": "a", "init_cost": 1.0}]} + + @pytest.mark.parametrize( + "fmt, parse", [("yaml", yaml.safe_load), ("json", json.loads)] + ) + def test_writes_to_dict_output(self, tmp_path, fmt, parse): + path = tmp_path / f"measures.{fmt}" + with patch.object(MeasureSet, "to_dict", return_value=self.PAYLOAD): + getattr(MeasureSet([]), f"to_{fmt}")(path) + + assert parse(path.read_text(encoding="utf-8")) == self.PAYLOAD + + @pytest.mark.parametrize("fmt, dump", [("yaml", yaml.dump), ("json", json.dumps)]) + def test_reads_and_forwards_to_from_dict(self, tmp_path, fmt, dump): + path = tmp_path / f"measures.{fmt}" + path.write_text(dump(self.PAYLOAD), encoding="utf-8") + + with patch.object(MeasureSet, "from_dict") as mock_from_dict: + result = getattr(MeasureSet, f"from_{fmt}")(path) + + mock_from_dict.assert_called_once_with(self.PAYLOAD) + assert result is mock_from_dict.return_value diff --git a/climada/util/string_parsers.py b/climada/util/string_parsers.py new file mode 100644 index 000000000..1184b5c89 --- /dev/null +++ b/climada/util/string_parsers.py @@ -0,0 +1,68 @@ +""" +This file is part of CLIMADA. + +Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS. + +CLIMADA is free software: you can redistribute it and/or modify it under the +terms of the GNU General Public License as published by the Free +Software Foundation, version 3. + +CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with CLIMADA. If not, see . + +--- + +Define functions to parse strings +""" + +from itertools import chain + +import pandas as pd + + +def parse_mapping_string(map_str: str | None) -> dict[int, int] | None: + """ + Parses strings like "1to8, 5to2" into {1: 8, 5: 2}. + """ + if map_str is None or map_str == "" or map_str == "nil": + return None + # 1. Split by comma to get individual pairs: ['1to8', '5to2'] + pairs = (p.strip() for p in map_str.split(",") if p.strip()) + + # 2. Split each pair by 'to' and convert to integers + def split_pair(p): + k, v = p.split("to") + return int(k), int(v) + + # 3. Construct the dictionary + return dict(split_pair(p) for p in pairs) + + +def parse_range(s: str | None) -> list | None: + """ + Parses strings like "1,4-6,10,12-14" into [1,4,5,6,10,12,13,14]. + """ + if s is None or s == "nil" or s == "": + return None + try: + return list( + chain.from_iterable( + range(r[0], r[-1] + 1) + for r in [[int(i) for i in part.split("-")] for part in s.split(",")] + ) + ) + except ValueError as exc: + raise ValueError( + f"Invalid string format for zeroing assets: {s} (Ex: '1,4-6,10')" + ) from exc + + +def parse_color(color_value: str | None) -> tuple | None: + """Parse RGB color string to tuple.""" + if color_value is not None and not pd.isna(color_value): + return tuple(color_value.split(" ")) + return None diff --git a/doc/user-guide/climada_measures.ipynb b/doc/user-guide/climada_measures.ipynb index d15ad4282..735fcdf9c 100644 --- a/doc/user-guide/climada_measures.ipynb +++ b/doc/user-guide/climada_measures.ipynb @@ -17,7 +17,9 @@ "source": [ "The `Measure` object is the core component of the **Adaptation** module in CLIMADA. It acts as a wrapper for transformation logic, allowing you to model how specific interventions alter the risk landscape (like building sea walls, retrofitting houses, or implementing early warning systems).\n", "\n", - "The primary purpose of a `Measure` is to modify one or more parts of the \"Risk Triplet\": **Exposures**, **Impact Functions**, and **Hazard**." + "The primary purpose of a `Measure` is to modify one or more parts of the \"Risk Triplet\": **Exposures**, **Impact Functions**, and **Hazard**.\n", + "\n", + "`Measure` objects can be regrouped into a set with the `MeasureSet` object." ] }, { @@ -28,42 +30,59 @@ "## Quickstart" ] }, + { + "cell_type": "markdown", + "id": "d369e8bd-1f00-4435-9b35-ef03bba2af97", + "metadata": {}, + "source": [ + "Assume you want to model two measures as follow:\n", + "- The first one is a targeted awareness campaign, which effect is modelled as a change of one specific impact function. Costing 500'000.\n", + "- The second one is both building a sea wall which effect is modelled as a 0.2 meter reduction of the flood height everywhere, and a building retrofit, modelled as an overal reduction of the value at risk of the exposure. Costing 1M." + ] + }, { "cell_type": "code", - "execution_count": null, + "execution_count": 1, "id": "2c48b021", "metadata": {}, "outputs": [], "source": [ "from climada.entity.measures.measure import Measure\n", + "from climada.entity.measures.measure_set import MeasureSet\n", "from climada.entity.measures.cost_income import CostIncome\n", "\n", "\n", - "# Define one or more transformation function\n", - "# It must take an Exposures, Hazard or ImpactFuncSEt as the first argument\n", - "# and return an object of the same type.\n", - "def reduce_exposure_value_at_risk(exposures, reduction_factor=0.2):\n", + "# Define the effect with python functions\n", + "def targeted_campaign(impact_func_set, impf_id, rescale):\n", + " \"\"\"Rescales the impact function with id impf_id\"\"\"\n", + " impact_func_set.get_func(fun_id=imp_id).mdd *= rescale\n", + " return impact_func_set\n", + "\n", + "\n", + "def building_retrofit(exposures, reduction_factor):\n", " \"\"\"Reduces the value of all assets by a fixed percentage.\"\"\"\n", " exposures.gdf[\"value\"] *= 1 - reduction_factor\n", " return exposures\n", "\n", "\n", - "def sea_wall_protection(hazard, threshold=2.0):\n", - " \"\"\"Cuts off flood intensity below a certain threshold.\"\"\"\n", - " # intensity is typically a sparse matrix in CLIMADA hazards\n", - " hazard.intensity.data[hazard.intensity.data < threshold] = 0\n", + "def sea_wall_protection(hazard, reduction):\n", + " \"\"\"Reduces flood intensity by an absolute amount.\"\"\"\n", + " hazard.intensity.data -= reduction\n", " return hazard\n", "\n", "\n", - "# Define financial aspects\n", - "cost_income = CostIncome(init_cost=1_000_000)\n", + "# Initialize the Measure objects\n", + "campaign = Measure(\n", + " name=\"Targeted awareness campaign\",\n", + " impfset_changes=targeted_campaign,\n", + " cost_income=CostIncome(init_cost=500_000),\n", + ")\n", "\n", - "# Initialize the Measure object\n", "retrofitting_and_seawall = Measure(\n", " name=\"Building Retrofit and seawall construction\",\n", - " exposures_changes=reduce_exposure_value_at_risk,\n", + " exposures_changes=building_retrofit,\n", " hazard_changes=sea_wall_protection,\n", - " cost_income=cost_income,\n", + " cost_income=CostIncome(init_cost=1_000_000),\n", ")" ] }, @@ -73,13 +92,20 @@ "metadata": {}, "source": [ "The measure can then be used either to create a new triplet corresponding to the situation with measure in effect\n", - "or to compute the impact directly (both lines are independant):\n", + "or to compute the impact directly (both lines are independent):\n", "\n", "```python\n", - "new_exp, new_impf, new_haz = retrofitting_and_seawall.apply(exp, impf, haz, kwargs_hazard={'threshold': 3.5})\n", + "new_exp, new_impf, new_haz = retrofitting_and_seawall.apply(exp, impf, haz, kwargs_exposure={'reduction_factor': 0.1}, kwargs_hazard={'reduction': 0.3})\n", "\n", "# or directly:\n", - "impact_with_measure = retrofitting_and_seawall.calc_impact(exp, impf, haz, kwargs_hazard={'threshold': 3.5})\n", + "impact_with_campaign_active = campaign.calc_impact(exp, impf, haz, kwargs_impfset={'rescale': 0.2})\n", + "```\n", + "\n", + "Much like impact functions, the measure can also be regrouped into a set with `MeasureSet`:\n", + "\n", + "```python\n", + "\n", + "meas_set = MeasureSet([campaign, retrofitting_and_seawall])\n", "```" ] }, @@ -111,7 +137,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 2, "id": "7cd53344", "metadata": {}, "outputs": [], @@ -256,7 +282,7 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 3, "id": "c3d90b57", "metadata": {}, "outputs": [], @@ -321,7 +347,7 @@ }, { "cell_type": "code", - "execution_count": 3, + "execution_count": 4, "id": "7e17dfb9-b760-48df-8878-be1d547290a3", "metadata": {}, "outputs": [], @@ -452,7 +478,7 @@ { "cell_type": "code", "execution_count": null, - "id": "fabfd29c-0b5e-442a-912f-3cbfac45f78d", + "id": "2bcbb4ec-701d-4e38-9321-e86a7dea9538", "metadata": {}, "outputs": [], "source": [ @@ -467,8 +493,15 @@ " \"relocation_and_standard\",\n", " exposures_changes=relocate_assets,\n", " hazard_changes=protection_standard,\n", - ")\n", - "\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "4e2ea806-8bca-4681-8543-750b9b303a59", + "metadata": {}, + "source": [ + "```python\n", "exp_relocated, _, _ = meas_relocation.apply(\n", " exp,\n", " impfset,\n", @@ -486,7 +519,136 @@ " year=2030, start_year=2025, full_year=2045, fraction=0.3, regions=[1, 2]\n", " ),\n", " kwargs_hazard=dict(year=2030, start_year=2025, full_year=2045, threshold=100000),\n", - ")" + ")\n", + "```" + ] + }, + { + "cell_type": "markdown", + "id": "fabfd29c-0b5e-442a-912f-3cbfac45f78d", + "metadata": {}, + "source": [ + "## Using `MeasureSet` objects\n", + "\n", + "`MeasureSet` are used to group a set of measure together, notably to compare them in a cost benefit analysis.\n", + "\n", + "`MeasureSet` are build from a list of measures, taking the example from the quickstart:" + ] + }, + { + "cell_type": "code", + "execution_count": 6, + "id": "04e747b4-f856-43c3-ba43-a1878c112afe", + "metadata": {}, + "outputs": [], + "source": [ + "from climada.entity.measures.measure import Measure\n", + "from climada.entity.measures.measure_set import MeasureSet\n", + "from climada.entity.measures.cost_income import CostIncome\n", + "\n", + "\n", + "# Define the effect with python functions\n", + "def targeted_campaign(impact_func_set, impf_id, rescale):\n", + " \"\"\"Rescales the impact function with id impf_id\"\"\"\n", + " impact_func_set.get_func(fun_id=imp_id).mdd *= rescale\n", + " return impact_func_set\n", + "\n", + "\n", + "def building_retrofit(exposures, reduction_factor):\n", + " \"\"\"Reduces the value of all assets by a fixed percentage.\"\"\"\n", + " exposures.gdf[\"value\"] *= 1 - reduction_factor\n", + " return exposures\n", + "\n", + "\n", + "def sea_wall_protection(hazard, reduction):\n", + " \"\"\"Reduces flood intensity by an absolute amount.\"\"\"\n", + " hazard.intensity.data -= reduction\n", + " return hazard\n", + "\n", + "\n", + "# Initialize the Measure objects\n", + "campaign = Measure(\n", + " name=\"Targeted awareness campaign\",\n", + " impfset_changes=targeted_campaign,\n", + " cost_income=CostIncome(init_cost=500_000),\n", + ")\n", + "\n", + "retrofitting_and_seawall = Measure(\n", + " name=\"Building Retrofit and seawall construction\",\n", + " exposures_changes=building_retrofit,\n", + " hazard_changes=sea_wall_protection,\n", + " cost_income=CostIncome(init_cost=1_000_000),\n", + ")\n", + "\n", + "meas_set = MeasureSet([campaign, retrofitting_and_seawall])" + ] + }, + { + "cell_type": "markdown", + "id": "024822ed-53fc-406e-a9f0-012b244113c8", + "metadata": {}, + "source": [ + "You can list the measures names of a set, get its size, check if it contains a measure (by name), `append()` an additional measure, `select()` them by their name." + ] + }, + { + "cell_type": "code", + "execution_count": 7, + "id": "2d532d22-6b74-45a4-8b14-3e7b9ede2def", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "There are 2 measures in the set:\n", + " ['Targeted awareness campaign', 'Building Retrofit and seawall construction']\n", + "True\n", + "False\n" + ] + } + ], + "source": [ + "print(f\"There are {len(meas_set)} measures in the set:\\n\", meas_set.names)\n", + "print(\"Targeted awareness campaign\" in meas_set)\n", + "print(\"Mangroves\" in meas_set)" + ] + }, + { + "cell_type": "code", + "execution_count": 8, + "id": "cdfc53b8-5da4-4b72-8fa5-7a7f5e18a6ea", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "{'Targeted awareness campaign': }" + ] + }, + "execution_count": 8, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "meas_set.select(\"Targeted awareness campaign\")" + ] + }, + { + "cell_type": "markdown", + "id": "8a29d4b4-983f-4fbd-894b-01716c899dfc", + "metadata": {}, + "source": [ + "You can also combine or compose measures of a `MeasureSet` into one `Measure`.\n", + "\n", + "1. *Combining* (`.combine()`) merges the effects of all the measures, taking the best possible effect when multiple measures affect the same dimension.\n", + "\n", + "For instance:\n", + "- When several measures of the set reduce the intensity of the hazard, the combined effect will be the point-wise highest reduction across all measures.\n", + "- When several measures modify the same impact function, the combined measure uses the point-wise minimum of all the modified impact functions.\n", + "\n", + "2. *Composing* (`.compose()`) does the mathematical composition of the measures, i.e., for a list of measures `[m1, m2, m3]`, the resulting transformation is `m1(m2(m3(x)))`. Note that this means measure `m3` is applied first, and the order of list of measures given to the `MeasureSet` matters!" ] } ],