diff --git a/CHANGELOG.md b/CHANGELOG.md index a77a16420..77db657fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ Code freeze date: YYYY-MM-DD ### Added - Better type hints and overloads signatures for ImpactFuncSet [#1250](https://github.com/CLIMADA-project/climada_python/pull/1250) +- Adds `MeasureConfig` and related dataclasses for new `Measure` object retrocompatibility and (de)serialization capabilities [#1276](https://github.com/CLIMADA-project/climada_python/pull/1276) - Add inter- and extrapolation options to `ImpactFreqCurve` with method `interpolate` [#1252](https://github.com/CLIMADA-project/climada_python/pull/1252) ### Changed diff --git a/climada/entity/measures/__init__.py b/climada/entity/measures/__init__.py new file mode 100644 index 000000000..f536fabfa --- /dev/null +++ b/climada/entity/measures/__init__.py @@ -0,0 +1,27 @@ +""" +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 . + +--- + +This module implements measure and measure set objects, as well as cost income +objects which enable the definition of adapation measures and their associated +effects on each part of risk/impacts (exposure, vulnerability and hazard). + +""" + +from .measure_config import MeasureConfig + +__all__ = ["MeasureConfig"] diff --git a/climada/entity/measures/measure_config.py b/climada/entity/measures/measure_config.py new file mode 100644 index 000000000..7b63c728b --- /dev/null +++ b/climada/entity/measures/measure_config.py @@ -0,0 +1,648 @@ +""" +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 configuration dataclasses for Measure reading and writing. +""" + +from __future__ import annotations + +import logging +import warnings +from dataclasses import MISSING, dataclass, field, fields +from datetime import datetime +from typing import Any, Dict, Optional, Tuple + +import numpy as np +import pandas as pd +import yaml + +LOGGER = logging.getLogger(__name__) + +__all__ = [ + "HazardModifierConfig", + "ExposuresModifierConfig", + "ImpfsetModifierConfig", + "CostIncomeConfig", + "MeasureConfig", +] + + +@dataclass +class ModifierConfig: + """ + Abstract base class for all modifier configuration dataclasses. + + Provides shared serialization, deserialization, and representation + logic for all concrete modifier config subclasses. Not intended to + be instantiated directly. + """ + + def _filter_out_default_fields(self) -> tuple[dict[str, Any], dict[str, Any]]: + """ + Partition the instance's fields into non-default and default groups. + + The ``haz_type`` field is always excluded from the output, as it + is managed at the ``MeasureConfig`` level. + + Returns + ------- + non_defaults : dict + Fields whose current value differs from the dataclass default. + defaults : dict + Fields whose current value equals the dataclass default. + """ + + non_defaults = {} + defaults = {} + for defined_field in fields(self): + val = getattr(self, defined_field.name) + default = defined_field.default + if defined_field.default_factory is not MISSING: + default = defined_field.default_factory() + + if val != default: + non_defaults[defined_field.name] = val + else: + defaults[defined_field.name] = val + + if "haz_type" in non_defaults: + non_defaults.pop("haz_type") + return non_defaults, defaults + + def to_dict(self, omit_default: bool = True) -> dict[str, Any]: + """ + Serialize the config to a flat dictionary, omitting default values. + + The ``haz_type`` field is always excluded from the output, as it + is managed at the ``MeasureConfig`` level. + + Returns + ------- + dict + Dictionary containing only fields whose values differ from + their dataclass defaults. + """ + non_defaults, defaults = self._filter_out_default_fields() + if omit_default: + return non_defaults + + return defaults | non_defaults + + @classmethod + def from_dict(cls, kwargs_dict: dict): + """ + Instantiate a config from a dictionary, ignoring unknown keys. + + Parameters + ---------- + kwargs_dict : dict + Input dictionary. Keys not matching any dataclass field are + silently discarded. + + Returns + ------- + _ModifierConfig + A new instance of the calling subclass. + """ + + filtered = cls._filter_dict_to_fields(kwargs_dict) + return cls(**filtered) + + @classmethod + def _filter_dict_to_fields(cls, to_filter: dict): + """ + Filter a dictionary to only the keys matching the dataclass fields. + + Parameters + ---------- + to_filter : dict + Input dictionary, potentially containing extra keys. + + Returns + ------- + dict + A copy of ``to_filter`` restricted to keys that correspond to declared + dataclass fields on this class. + """ + + field_names = [f.name for f in fields(cls)] + return {key: val for key, val in to_filter.items() if key in field_names} + + def __repr__(self) -> str: + """ + Return a human-readable representation highlighting non-default fields. + + Non-default fields are shown prominently; default fields are shown + below them. This makes it easy to see at a glance what has been + configured on an instance. + + Returns + ------- + str + A formatted string representation of the instance. + """ + + non_defaults, defaults = self._filter_out_default_fields() + ndf_fields_str = ( + "\n\t\t\t".join(f"{k}={v!r}" for k, v in non_defaults.items()) + if non_defaults + else None + ) + _ = ( + "\n\t\t\t".join(f"{k}={v!r}" for k, v in defaults.items()) + if defaults + else None + ) + ndf_fields = ( + "(" "\n\t\tNon default fields:" f"\n\t\t\t{ndf_fields_str}" "\n)" + if ndf_fields_str + else "()" + ) + return f"{self.__class__.__name__}{ndf_fields}" + + +@dataclass(repr=False) +class ImpfsetModifierConfig(ModifierConfig): + """ + Configuration for modifications to an impact function set. + + Supports scaling or shifting MDD, PAA, and intensity curves, as well + as replacement of the impact function set, loaded from a file path. If + both a new file path and modifier values are provided, modifiers are + applied after the replacement (and a warning is issued). + + Parameters + ---------- + haz_type : str + Hazard type identifier (e.g. ``"TC"``) that this modifier targets. + impf_ids : int or str or list of int or str, optional + Impact function ID(s) to which modifications are applied. + If ``None``, all impact functions are affected. + impf_mdd_mult : float, optional + Multiplicative factor applied to the mean damage degree (MDD) curve. + Default is ``1.0`` (no change). + impf_mdd_add : float, optional + Additive offset applied to the MDD curve after multiplication. + Default is ``0.0``. + impf_paa_mult : float, optional + Multiplicative factor applied to the percentage of affected assets + (PAA) curve. Default is ``1.0``. + impf_paa_add : float, optional + Additive offset applied to the PAA curve after multiplication. + Default is ``0.0``. + impf_int_mult : float, optional + Multiplicative factor applied to the intensity axis. + Default is ``1.0``. + impf_int_add : float, optional + Additive offset applied to the intensity axis after multiplication. + Default is ``0.0``. + new_impfset_path : str, optional + Path to an Excel file containing a replacement impact function set. + If provided alongside modifier values, a warning is issued and + modifiers are applied after loading the new set. + + Warns + ----- + UserWarning + If ``new_impfset_path`` is set alongside any non-default modifier + values. + """ + + haz_type: str + impf_ids: int | str | list[int | str] | None = None + impf_mdd_mult: float = 1.0 + impf_mdd_add: float = 0.0 + impf_paa_mult: float = 1.0 + impf_paa_add: float = 0.0 + impf_int_mult: float = 1.0 + impf_int_add: float = 0.0 + new_impfset_path: str | None = None + + def __post_init__(self): + config = self.to_dict() + if "new_impfset_path" in config and any( + key in config + for key in [ + "impf_mdd_add", + "impf_mdd_mult", + "impf_paa_add", + "impf_paa_mult", + "impf_int_add", + "impf_int_mult", + ] + ): + warnings.warn( + "Both new impfset object and impfset modifiers are provided, " + "modifiers will be applied after changing the impfset." + ) + + +@dataclass(repr=False) +class HazardModifierConfig(ModifierConfig): + """ + Configuration for modifications to a hazard. + + Supports scaling or shifting hazard intensity, applying a return-period + frequency cutoff, and replacement of the hazard, loaded from a file path. + If both a new file path and modifier values are provided, modifiers are + applied after the replacement. + + Parameters + ---------- + haz_type : str + Hazard type identifier (e.g. ``"TC"``) that this modifier targets. + haz_int_mult : float, optional + Multiplicative factor applied to hazard intensity. + Default is ``1.0`` (no change). + haz_int_add : float, optional + Additive offset applied to hazard intensity after multiplication. + Default is ``0.0``. + new_hazard_path : str, optional + Path to an HDF5 file containing a replacement hazard. + If provided alongside modifier values, a warning is issued and + modifiers are applied after loading the new hazard. + impact_rp_cutoff : float, optional + Return period (in years) below which hazard events are discarded. + If ``None``, no cutoff is applied. + + Warns + ----- + UserWarning + If ``new_hazard_path`` is set alongside any non-default modifier + values or a non-``None`` ``impact_rp_cutoff``. + """ + + haz_type: str + haz_int_mult: Optional[float] = 1.0 + haz_int_add: Optional[float] = 0.0 + haz_freq_mult: Optional[float] = 1.0 + haz_freq_add: Optional[float] = 0.0 + new_hazard_path: Optional[str] = None + impact_rp_cutoff: Optional[float] = None + + def __post_init__(self): + config = self.to_dict() + if "new_hazard_path" in config and any( + key in config + for key in [ + "haz_int_mult", + "haz_int_add", + "haz_freq_mult", + "haz_freq_add", + "impact_rp_cutoff", + ] + ): + warnings.warn( + "Both new hazard object and hazard modifiers are provided, " + "modifiers will be applied after changing the hazard." + ) + + +@dataclass(repr=False) +class ExposuresModifierConfig(ModifierConfig): + """ + Configuration for modifications to an exposures object. + + Supports remapping impact function IDs, zeroing out selected regions, + and replacement of the exposures from a new file. If both a new + file path and modifier values are provided, modifiers are applied after + the replacement. + + Parameters + ---------- + reassign_impf_id : dict of {str: dict of {int or str: int or str}}, optional + Nested mapping ``{haz_type: {old_id: new_id}}`` used to reassign + impact function IDs in the exposures. If ``None``, no remapping + is performed. + set_to_zero : list of int, optional + Region IDs for which exposure values are set to zero. + If ``None``, no zeroing is applied. + new_exposures_path : str, optional + Path to an HDF5 file containing replacement exposures. + If provided alongside modifier values, a warning is issued and + modifiers are applied after loading the new exposures. + + Warns + ----- + UserWarning + If ``new_exposures_path`` is set alongside any non-``None`` + modifier values. + """ + + reassign_impf_id: Optional[Dict[str, Dict[int | str, int | str]]] = None + set_to_zero: Optional[list[int]] = None + new_exposures_path: Optional[str] = None + + def __post_init__(self): + config = self.to_dict() + if "new_exposures_path" in config and any( + key in config for key in ["reassign_impf_id", "set_to_zero"] + ): + warnings.warn( + "Both new exposures object and exposures modifiers are provided, " + "modifiers will be applied after changing the exposures." + ) + + +@dataclass(repr=False) +class CostIncomeConfig(ModifierConfig): + """ + Serializable configuration for a ``CostIncome`` object. + + Encodes all parameters required to construct a ``CostIncome`` instance, + including optional custom cash flow schedules. + + Parameters + ---------- + mkt_price_year : int, optional + Reference year for market prices. Defaults to the current year. + init_cost : float, optional + One-time initial investment cost (positive value). Default is ``0.0``. + periodic_cost : float, optional + Recurring cost per period (positive value). Default is ``0.0``. + periodic_income : float, optional + Recurring income per period. Default is ``0.0``. + cost_yearly_growth_rate : float, optional + Annual growth rate applied to periodic costs. Default is ``0.0``. + income_yearly_growth_rate : float, optional + Annual growth rate applied to periodic income. Default is ``0.0``. + freq : str, optional + Pandas period alias defining the period length (e.g. ``"Y"`` for + yearly, ``"M"`` for monthly). Default is ``"Y"``. + See [pandas documentation](https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#period-aliases). + custom_cash_flows : list of dict, optional + Explicit cash flow schedule as a list of records with at minimum + a ``"date"`` key (ISO 8601 string) and a value key. If provided, + overrides the periodic cost/income logic. + """ # noqa + + mkt_price_year: Optional[int] = field(default_factory=lambda: datetime.today().year) + init_cost: float = 0.0 + periodic_cost: float = 0.0 + periodic_income: float = 0.0 + cost_yearly_growth_rate: float = 0.0 + income_yearly_growth_rate: float = 0.0 + freq: str = "Y" + custom_cash_flows: Optional[list[dict]] = None + + @classmethod + def from_cost_income(cls, cost_income: "CostIncome") -> "CostIncomeConfig": + """ + Construct a :class:`CostIncomeConfig` from a live + :class:`CostIncome` object. + + Parameters + ---------- + cost_income : CostIncome + The live ``CostIncome`` instance to serialise. + + Returns + ------- + CostIncomeConfig + The config instance equivalent to the ``CostIncome``. + """ + + custom = None + if cost_income.custom_cash_flows is not None: + custom = ( + cost_income.custom_cash_flows.reset_index() + .rename(columns={"index": "date"}) + .assign(date=lambda df: df["date"].dt.strftime("%Y-%m-%d")) + .to_dict(orient="records") + ) + return cls( + mkt_price_year=cost_income.mkt_price_year.year, # datetime → int + init_cost=abs(cost_income.init_cost), # stored negative → positive + periodic_cost=abs(cost_income.periodic_cost), + periodic_income=cost_income.periodic_income, + cost_yearly_growth_rate=cost_income.cost_growth_rate, + income_yearly_growth_rate=cost_income.income_growth_rate, + freq=cost_income.freq, + custom_cash_flows=custom, + ) + + +@dataclass(repr=False) +class MeasureConfig(ModifierConfig): + """ + Top-level serializable configuration for a single adaptation measure. + + Aggregates all modifier sub-configs (hazard, impact functions, exposures, + cost/income) into a single object that can be round-tripped through dict, + YAML, or a legacy Excel row. + + This class is the primary entry point for defining measures in a + declarative, file-based workflow and serves as the serialization + counterpart to :class:`~climada.entity.measures.base.Measure`. + + Parameters + ---------- + name : str + Unique name identifying this measure. + haz_type : str + Hazard type identifier (e.g. ``"TC"``) this measure is designed for. + impfset_modifier : ImpfsetModifierConfig + Configuration describing modifications to the impact function set. + hazard_modifier : HazardModifierConfig + Configuration describing modifications to the hazard. + exposures_modifier : ExposuresModifierConfig + Configuration describing modifications to the exposures. + cost_income : CostIncomeConfig + Financial parameters associated with implementing this measure. + implementation_duration : str, optional + Pandas period alias (e.g. ``"2Y"``) representing the time before + the measure is fully operational. If ``None``, the measure takes + effect immediately. + color_rgb : tuple of float, optional + RGB colour triple in the range ``[0, 1]`` used for visualisation. + If ``None``, defaults to black ``(0, 0, 0)``. + """ + + name: str + haz_type: str + impfset_modifier: ImpfsetModifierConfig + hazard_modifier: HazardModifierConfig + exposures_modifier: ExposuresModifierConfig + cost_income: CostIncomeConfig + implementation_duration: Optional[str] = None + color_rgb: Optional[Tuple[float, float, float]] = None + + def __repr__(self) -> str: + """ + Return a detailed string representation of the measure configuration. + + All fields are shown, including sub-configs, with each on its own + indented line. + + Returns + ------- + str + A formatted multi-line string representation. + """ + + fields_str = "\n\t".join(f"{k}={v!r}" for k, v in self.__dict__.items()) + return f"{self.__class__.__name__}(\n\t{fields_str})" + + def to_dict(self, omit_default: bool = True) -> dict: + """ + Serialize the measure configuration to a flat dictionary. + + Sub-config dictionaries are merged into the top-level dict (i.e. + their keys are inlined, not nested). ``haz_type`` is always included + at the top level. Fields with ``None`` values are preserved. + + Returns + ------- + dict + Flat dictionary representation suitable for YAML or Excel + serialization. + """ + + return { + "name": self.name, + "haz_type": self.haz_type, + **self.impfset_modifier.to_dict(omit_default), + **self.hazard_modifier.to_dict(omit_default), + **self.exposures_modifier.to_dict(omit_default), + **self.cost_income.to_dict(omit_default), + "implementation_duration": self.implementation_duration, + "color_rgb": list(self.color_rgb) if self.color_rgb is not None else None, + } + + @classmethod + def from_dict(cls, kwargs_dict: dict) -> "MeasureConfig": + """ + Instantiate a :class:`MeasureConfig` from a flat dictionary. + + Delegates sub-config construction to the respective + ``from_dict`` classmethods. Unknown keys are silently discarded + by each sub-config parser. + + Parameters + ---------- + kwargs_dict : dict + Flat dictionary, as produced by :meth:`to_dict` or read from + a legacy Excel row. Must contain at minimum ``"name"`` and + ``"haz_type"``. + + Returns + ------- + MeasureConfig + A fully populated configuration instance. + """ + + return cls( + name=kwargs_dict["name"], + haz_type=kwargs_dict["haz_type"], + impfset_modifier=ImpfsetModifierConfig.from_dict(kwargs_dict), + hazard_modifier=HazardModifierConfig.from_dict(kwargs_dict), + exposures_modifier=ExposuresModifierConfig.from_dict(kwargs_dict), + cost_income=CostIncomeConfig.from_dict(kwargs_dict), + implementation_duration=kwargs_dict.get("implementation_duration"), + color_rgb=cls._normalize_color(kwargs_dict.get("color_rgb")), + ) + + @staticmethod + def _normalize_color(color_rgb): + # 1. Handle None and NaN (np.nan, pd.NA, float('nan')) + if color_rgb is None or pd.isna(color_rgb) is True: + return None + + # 2. Convert sequence types (list, np.array, tuple) to a standard tuple + try: + # Flatten in case it's a nested numpy array, then convert to tuple + result = tuple(np.array(color_rgb).flatten().tolist()) + + # 3. Enforce the length of three + if len(result) != 3: + raise ValueError(f"Expected 3 digits, got {len(result)}") + + return result + + except (TypeError, ValueError) as err: + # Handle cases where input isn't iterable or wrong length + raise ValueError(f"Invalid color format: {color_rgb}.") from err + + def to_yaml(self, path: str) -> None: + """ + Write this configuration to a YAML file. + + The file is structured as ``{"measures": []}``, + matching the expected format for :meth:`from_yaml`. + + Parameters + ---------- + path : str + Destination file path. Will be created or overwritten. + """ + + with open(path, "w") as opened_file: + yaml.dump( + {"measures": [self.to_dict()]}, + opened_file, + default_flow_style=False, + sort_keys=False, + ) + + @classmethod + def from_yaml(cls, path: str) -> "MeasureConfig": + """ + Load a :class:`MeasureConfig` from a YAML file. + + Expects the file to contain a top-level ``"measures"`` list; reads + only the first entry. + + Parameters + ---------- + path : str + Path to the YAML file to read. + + Returns + ------- + MeasureConfig + The configuration parsed from the first entry in + ``measures``. + """ + + with open(path) as opened_file: + return cls.from_dict(yaml.safe_load(opened_file)["measures"][0]) + + @classmethod + def from_row(cls, row: pd.Series) -> "MeasureConfig": + """ + Construct a :class:`MeasureConfig` from a legacy Excel row. + + Converts the row to a dictionary and delegates to :meth:`from_dict`. + This is the primary migration path for measures currently stored in + the legacy Excel-based ``MeasureSet`` format. + + Parameters + ---------- + row : pd.Series + A single row from a legacy measures Excel sheet, with column + names matching the flat dictionary keys expected by + :meth:`from_dict`. + + Returns + ------- + MeasureConfig + A configuration instance populated from the row data. + """ + + row_dict = row.to_dict() + return cls.from_dict(row_dict) diff --git a/climada/entity/measures/test/test_measure_config.py b/climada/entity/measures/test/test_measure_config.py new file mode 100644 index 000000000..2fcd45113 --- /dev/null +++ b/climada/entity/measures/test/test_measure_config.py @@ -0,0 +1,425 @@ +""" +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 . + +--- + +Tests for MeasureConfig and related dataclasses. +""" + +# tests/entity/measures/test_measure_config.py + +import warnings +from datetime import datetime + +import pandas as pd +import pytest + +from climada.entity.measures.measure_config import ( + CostIncomeConfig, + ExposuresModifierConfig, + HazardModifierConfig, + ImpfsetModifierConfig, + MeasureConfig, +) + + +@pytest.fixture +def minimal_measure_dict(): + return {"name": "seawall", "haz_type": "TC"} + + +@pytest.fixture +def full_measure_dict(): + return { + "name": "seawall", + "haz_type": "TC", + "haz_int_mult": 0.8, + "haz_int_add": -0.1, + "impf_mdd_mult": 0.9, + "impf_paa_mult": 0.95, + "impf_ids": [1, 2], + "reassign_impf_id": {"TC": {1: 3}}, + "set_to_zero": [10, 20], + "init_cost": 1000.0, + "periodic_cost": 50.0, + "color_rgb": [0.1, 0.5, 0.9], + "implementation_duration": "2Y", + } + + +class TestModifierConfig: + + def test_to_dict_omits_defaults(self): + config = ImpfsetModifierConfig(haz_type="TC") + result = config.to_dict() + assert result == {} + + def test_to_dict_includes_non_defaults_no_omit(self): + config = ImpfsetModifierConfig( + haz_type="TC", impf_mdd_mult=0.5, impf_paa_add=0.1 + ) + result = config.to_dict(omit_default=False) + assert sorted(list(result.keys())) != sorted( + ["haz_type", "impf_mdd_mult", "impf_paa_add"] + ) + assert result["impf_mdd_mult"] == 0.5 + assert result["impf_paa_add"] == 0.1 + + def test_to_dict_includes_non_defaults(self): + config = ImpfsetModifierConfig( + haz_type="TC", impf_mdd_mult=0.5, impf_paa_add=0.1 + ) + result = config.to_dict() + assert sorted(list(result.keys())) == sorted(["impf_mdd_mult", "impf_paa_add"]) + assert result["impf_mdd_mult"] == 0.5 + assert result["impf_paa_add"] == 0.1 + + def test_from_dict_ignores_unknown_keys(self): + d = {"haz_type": "TC", "unknown_field": 99, "another_unknown": "foo"} + config = ImpfsetModifierConfig.from_dict(d) + assert config.haz_type == "TC" + assert not hasattr(config, "unknown_field") + + def test_from_dict_roundtrip(self): + config = ImpfsetModifierConfig( + haz_type="TC", impf_mdd_mult=0.5, impf_paa_add=0.1 + ) + d = {**config.to_dict(), "haz_type": "TC"} + recovered = ImpfsetModifierConfig.from_dict(d) + assert recovered.impf_mdd_mult == config.impf_mdd_mult + assert recovered.impf_paa_add == config.impf_paa_add + + def test_repr_shows_non_defaults_prominently(self): + config = ImpfsetModifierConfig(haz_type="TC", impf_mdd_mult=0.5) + r = repr(config) + assert "Non default fields" in r + assert "impf_mdd_mult" in r + + def test_repr_empty_when_all_defaults(self): + config = ImpfsetModifierConfig(haz_type="TC") + r = repr(config) + assert "Non default fields" not in r + + +class TestImpfsetModifierConfig: + + def test_config_defaults(self): + config = ImpfsetModifierConfig(haz_type="TC") + assert config.impf_ids is None + assert config.impf_mdd_mult == 1.0 + assert config.impf_mdd_add == 0.0 + assert config.impf_paa_mult == 1.0 + assert config.impf_paa_add == 0.0 + assert config.impf_int_mult == 1.0 + assert config.impf_int_add == 0.0 + assert config.new_impfset_path is None + + def test_config_from_dict_roundtrip(self): + config = ImpfsetModifierConfig( + haz_type="TC", impf_mdd_mult=0.8, impf_ids=[1, 2] + ) + d = {**config.to_dict(), "haz_type": "TC"} + recovered = ImpfsetModifierConfig.from_dict(d) + assert recovered.impf_mdd_mult == config.impf_mdd_mult + assert recovered.impf_ids == config.impf_ids + + def test_config_to_dict_roundtrip(self): + d = {"haz_type": "TC", "impf_mdd_mult": 0.8, "impf_paa_add": 0.05} + config = ImpfsetModifierConfig.from_dict(d) + result = {**config.to_dict(), "haz_type": "TC"} + assert result["impf_mdd_mult"] == d["impf_mdd_mult"] + assert result["impf_paa_add"] == d["impf_paa_add"] + + def test_config_warns_when_path_and_modifiers_combined(self): + with pytest.warns(UserWarning): + ImpfsetModifierConfig( + haz_type="TC", + new_impfset_path="path/to/file.xlsx", + impf_mdd_mult=0.5, + ) + + def test_config_no_warning_when_only_path(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + ImpfsetModifierConfig(haz_type="TC", new_impfset_path="path/to/file.xlsx") + + def test_config_no_warning_when_only_modifiers(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + ImpfsetModifierConfig(haz_type="TC", impf_mdd_mult=0.5) + + def test_config_impf_ids_accepts_int(self): + config = ImpfsetModifierConfig(haz_type="TC", impf_ids=1) + assert config.impf_ids == 1 + + def test_config_impf_ids_accepts_str(self): + config = ImpfsetModifierConfig(haz_type="TC", impf_ids="1") + assert config.impf_ids == "1" + + def test_config_impf_ids_accepts_list(self): + config = ImpfsetModifierConfig(haz_type="TC", impf_ids=[1, 2, "3"]) + assert config.impf_ids == [1, 2, "3"] + + def test_config_impf_ids_accepts_none(self): + config = ImpfsetModifierConfig(haz_type="TC", impf_ids=None) + assert config.impf_ids is None + + +class TestHazardModifierConfig: + + def test_config_defaults(self): + config = HazardModifierConfig(haz_type="TC") + assert config.haz_int_mult == 1.0 + assert config.haz_int_add == 0.0 + assert config.new_hazard_path is None + assert config.impact_rp_cutoff is None + + def test_config_from_dict_roundtrip(self): + config = HazardModifierConfig(haz_type="TC", haz_int_mult=0.8, haz_int_add=-0.1) + d = {**config.to_dict(), "haz_type": "TC"} + recovered = HazardModifierConfig.from_dict(d) + assert recovered.haz_int_mult == config.haz_int_mult + assert recovered.haz_int_add == config.haz_int_add + + def test_config_to_dict_roundtrip(self): + d = {"haz_type": "TC", "haz_int_mult": 0.7, "haz_int_add": -0.2} + config = HazardModifierConfig.from_dict(d) + result = {**config.to_dict(), "haz_type": "TC"} + assert result["haz_int_mult"] == d["haz_int_mult"] + assert result["haz_int_add"] == d["haz_int_add"] + + def test_config_warns_when_path_and_modifiers_combined(self): + with pytest.warns(UserWarning): + HazardModifierConfig( + haz_type="TC", + new_hazard_path="path/to/hazard.h5", + haz_int_mult=0.5, + ) + + def test_config_warns_when_path_and_rp_cutoff_combined(self): + with pytest.warns(UserWarning): + HazardModifierConfig( + haz_type="TC", + new_hazard_path="path/to/hazard.h5", + impact_rp_cutoff=100.0, + ) + + def test_config_no_warning_when_only_path(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + HazardModifierConfig(haz_type="TC", new_hazard_path="path/to/hazard.h5") + + def test_config_no_warning_when_only_modifiers(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + HazardModifierConfig(haz_type="TC", haz_int_mult=0.5) + + +class TestExposuresModifierConfig: + + def test_config_defaults(self): + config = ExposuresModifierConfig() + assert config.reassign_impf_id is None + assert config.set_to_zero is None + assert config.new_exposures_path is None + + def test_config_from_dict_roundtrip(self): + config = ExposuresModifierConfig( + reassign_impf_id={"TC": {1: 2}}, + set_to_zero=[10, 20], + ) + d = config.to_dict() + recovered = ExposuresModifierConfig.from_dict(d) + assert recovered.reassign_impf_id == config.reassign_impf_id + assert recovered.set_to_zero == config.set_to_zero + + def test_config_to_dict_roundtrip(self): + d = {"reassign_impf_id": {"TC": {1: 2}}, "set_to_zero": [5, 6]} + config = ExposuresModifierConfig.from_dict(d) + result = config.to_dict() + assert result["reassign_impf_id"] == d["reassign_impf_id"] + assert result["set_to_zero"] == d["set_to_zero"] + + def test_config_warns_when_path_and_modifiers_combined(self): + with pytest.warns(UserWarning): + ExposuresModifierConfig( + new_exposures_path="path/to/exp.h5", + reassign_impf_id={"TC": {1: 2}}, + ) + + def test_config_no_warning_when_only_path(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + ExposuresModifierConfig(new_exposures_path="path/to/exp.h5") + + def test_config_no_warning_when_only_modifiers(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + ExposuresModifierConfig(reassign_impf_id={"TC": {1: 2}}) + + def test_config_reassign_impf_id_accepts_int_keys(self): + config = ExposuresModifierConfig(reassign_impf_id={"TC": {1: 2}}) + assert config.reassign_impf_id == {"TC": {1: 2}} + + def test_config_reassign_impf_id_accepts_str_keys(self): + config = ExposuresModifierConfig(reassign_impf_id={"TC": {"1": "2"}}) + assert config.reassign_impf_id == {"TC": {"1": "2"}} + + def test_config_set_to_zero_accepts_none(self): + config = ExposuresModifierConfig(set_to_zero=None) + assert config.set_to_zero is None + + def test_config_set_to_zero_accepts_list(self): + config = ExposuresModifierConfig(set_to_zero=[1, 2, 3]) + assert config.set_to_zero == [1, 2, 3] + + +class TestCostIncomeConfig: + + def test_config_defaults(self): + config = CostIncomeConfig() + assert config.init_cost == 0.0 + assert config.periodic_cost == 0.0 + assert config.periodic_income == 0.0 + assert config.cost_yearly_growth_rate == 0.0 + assert config.income_yearly_growth_rate == 0.0 + assert config.freq == "Y" + assert config.custom_cash_flows is None + + def test_config_default_mkt_price_year_is_current_year(self): + config = CostIncomeConfig() + assert config.mkt_price_year == datetime.today().year + + def test_config_from_dict_roundtrip(self): + config = CostIncomeConfig(init_cost=1000.0, periodic_cost=50.0, freq="M") + d = config.to_dict() + recovered = CostIncomeConfig.from_dict(d) + assert recovered.init_cost == config.init_cost + assert recovered.periodic_cost == config.periodic_cost + assert recovered.freq == config.freq + + def test_config_to_dict_roundtrip(self): + d = {"init_cost": 500.0, "periodic_income": 20.0, "freq": "M"} + config = CostIncomeConfig.from_dict(d) + result = config.to_dict() + assert result["init_cost"] == d["init_cost"] + assert result["periodic_income"] == d["periodic_income"] + assert result["freq"] == d["freq"] + + +class TestMeasureConfig: + + def test_from_dict_minimal(self, minimal_measure_dict): + config = MeasureConfig.from_dict(minimal_measure_dict) + assert config.name == "seawall" + assert config.haz_type == "TC" + assert config.impfset_modifier == ImpfsetModifierConfig(haz_type="TC") + assert config.hazard_modifier == HazardModifierConfig(haz_type="TC") + assert config.exposures_modifier == ExposuresModifierConfig() + assert config.cost_income == CostIncomeConfig() + + def test_from_dict_full(self, full_measure_dict): + config = MeasureConfig.from_dict(full_measure_dict) + assert config.hazard_modifier.haz_int_mult == full_measure_dict["haz_int_mult"] + assert ( + config.impfset_modifier.impf_mdd_mult == full_measure_dict["impf_mdd_mult"] + ) + assert config.exposures_modifier.set_to_zero == full_measure_dict["set_to_zero"] + assert config.cost_income.init_cost == full_measure_dict["init_cost"] + assert config.color_rgb == tuple(full_measure_dict["color_rgb"]) + assert ( + config.implementation_duration + == full_measure_dict["implementation_duration"] + ) + + def test_from_dict_ignores_unknown_keys(self, minimal_measure_dict): + d = {**minimal_measure_dict, "completely_unknown": 42} + config = MeasureConfig.from_dict(d) + assert config.name == "seawall" + assert not hasattr(config, "completely_unknown") + + def test_to_dict_roundtrip(self, full_measure_dict): + config = MeasureConfig.from_dict(full_measure_dict) + recovered = MeasureConfig.from_dict(config.to_dict()) + assert recovered.name == config.name + assert recovered.haz_type == config.haz_type + assert recovered.hazard_modifier == config.hazard_modifier + assert recovered.impfset_modifier == config.impfset_modifier + assert recovered.exposures_modifier == config.exposures_modifier + assert recovered.color_rgb == config.color_rgb + assert recovered.implementation_duration == config.implementation_duration + + def test_to_dict_color_rgb_none(self, minimal_measure_dict): + config = MeasureConfig.from_dict(minimal_measure_dict) + result = config.to_dict() + assert result["color_rgb"] is None + + def test_to_dict_color_rgb_set(self, minimal_measure_dict): + config = MeasureConfig.from_dict( + {**minimal_measure_dict, "color_rgb": [0.1, 0.5, 0.9]} + ) + result = config.to_dict() + assert result["color_rgb"] == [0.1, 0.5, 0.9] + + def test_to_yaml_roundtrip(self, tmp_path, full_measure_dict): + path = str(tmp_path / "measure.yaml") + config = MeasureConfig.from_dict(full_measure_dict) + config.to_yaml(path) + recovered = MeasureConfig.from_yaml(path) + assert recovered.name == config.name + assert recovered.haz_type == config.haz_type + assert recovered.hazard_modifier == config.hazard_modifier + assert recovered.impfset_modifier == config.impfset_modifier + assert recovered.color_rgb == config.color_rgb + + def test_from_yaml_reads_first_entry(self, tmp_path, full_measure_dict): + import yaml + + second = {**full_measure_dict, "name": "second_measure"} + path = str(tmp_path / "measures.yaml") + with open(path, "w") as f: + yaml.dump({"measures": [full_measure_dict, second]}, f) + config = MeasureConfig.from_yaml(path) + assert config.name == full_measure_dict["name"] + + def test_from_row_roundtrip(self, full_measure_dict): + config = MeasureConfig.from_dict(full_measure_dict) + row = pd.Series(config.to_dict()) + recovered = MeasureConfig.from_row(row) + assert recovered.name == config.name + assert recovered.hazard_modifier == config.hazard_modifier + assert recovered.impfset_modifier == config.impfset_modifier + + def test_from_row_ignores_extra_columns(self, full_measure_dict): + config = MeasureConfig.from_dict(full_measure_dict) + d = {**config.to_dict(), "extra_column": "garbage"} + row = pd.Series(d) + recovered = MeasureConfig.from_row(row) + assert recovered.name == config.name + + def test_sub_configs_correctly_dispatched(self, full_measure_dict): + config = MeasureConfig.from_dict(full_measure_dict) + assert config.hazard_modifier.haz_int_mult == full_measure_dict["haz_int_mult"] + assert ( + config.impfset_modifier.impf_mdd_mult == full_measure_dict["impf_mdd_mult"] + ) + assert ( + config.exposures_modifier.reassign_impf_id + == full_measure_dict["reassign_impf_id"] + ) + assert config.cost_income.init_cost == full_measure_dict["init_cost"] + assert not hasattr(config.hazard_modifier, "impf_mdd_mult") + assert not hasattr(config.impfset_modifier, "haz_int_mult") diff --git a/doc/api/climada/climada.entity.measures.rst b/doc/api/climada/climada.entity.measures.rst new file mode 100644 index 000000000..62f8b2d94 --- /dev/null +++ b/doc/api/climada/climada.entity.measures.rst @@ -0,0 +1,14 @@ +climada\.entity\.measures package +================================= + +.. note:: + This package implements the new way of defining measures. + For the previous way, see :ref:`climada.entity._legacy_measures` + +climada\.entity\.measures\.measure_config module +------------------------------------------------ + +.. automodule:: climada.entity.measures.measure_config + :members: + :undoc-members: + :show-inheritance: diff --git a/doc/api/climada/climada.entity.rst b/doc/api/climada/climada.entity.rst index fc1c47920..f4f4df0d9 100644 --- a/doc/api/climada/climada.entity.rst +++ b/doc/api/climada/climada.entity.rst @@ -6,6 +6,7 @@ climada\.entity package climada.entity.disc_rates climada.entity.exposures climada.entity.impact_funcs + climada.entity.measures climada.entity._legacy_measures climada\.entity\.entity\_def module diff --git a/doc/user-guide/adaptation.rst b/doc/user-guide/adaptation.rst index 1ed8fb959..cd422bbcb 100644 --- a/doc/user-guide/adaptation.rst +++ b/doc/user-guide/adaptation.rst @@ -10,7 +10,7 @@ These guides show everything you need to know in order to evaluate adaptation op :maxdepth: 1 .. Adaptation measures in CLIMADA - .. Using measure configurations + Using measure configurations .. Defining measure cash flows .. Cost benefit evaluation .. Adapation planning evaluation diff --git a/doc/user-guide/climada_entity_Exposures.ipynb b/doc/user-guide/climada_entity_Exposures.ipynb index aa1b39fd3..90a0c81eb 100644 --- a/doc/user-guide/climada_entity_Exposures.ipynb +++ b/doc/user-guide/climada_entity_Exposures.ipynb @@ -4,6 +4,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ + "(exposure-tutorial)=\n", "# Exposures class" ] }, diff --git a/doc/user-guide/climada_entity_ImpactFuncSet.ipynb b/doc/user-guide/climada_entity_ImpactFuncSet.ipynb index fd349487c..ad1841a75 100644 --- a/doc/user-guide/climada_entity_ImpactFuncSet.ipynb +++ b/doc/user-guide/climada_entity_ImpactFuncSet.ipynb @@ -4,6 +4,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ + "(impact-functions-tutorial)=\n", "# Impact Functions" ] }, diff --git a/doc/user-guide/climada_hazard_Hazard.ipynb b/doc/user-guide/climada_hazard_Hazard.ipynb index 412346d04..0b6bd4037 100644 --- a/doc/user-guide/climada_hazard_Hazard.ipynb +++ b/doc/user-guide/climada_hazard_Hazard.ipynb @@ -4,6 +4,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ + "(hazard-tutorial)=\n", "# Hazard class\n", "\n", "## What is a hazard?\n", diff --git a/doc/user-guide/climada_measure_config.ipynb b/doc/user-guide/climada_measure_config.ipynb new file mode 100644 index 000000000..8fa4739ba --- /dev/null +++ b/doc/user-guide/climada_measure_config.ipynb @@ -0,0 +1,570 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "659605a5-d601-47d3-89f7-b606e3e39c93", + "metadata": {}, + "source": [ + "(measure-config-tutorial)=\n", + "\n", + "# Defining Adaptation Measures with configurations" + ] + }, + { + "cell_type": "markdown", + "id": "c68c6cf1-d0a1-40ae-ae45-838741988ac6", + "metadata": {}, + "source": [ + "## Introduction\n", + "\n", + "CLIMADA uses `Measure` objects to model the effects of adaptation measures. `Measure` objects were formerly defined declaratively (via for instance, a shifting or scaling of the hazard intensity or a change of impact function), and are now defined as python functions to enable more flexibility on the possible changes (see the [tutorial on measure objects](measure-tutorial)'). \n", + "\n", + "The caveat of defining measure effects as python functions is that it cannot be serialized (written to a file), and also makes reading from a file a challenge.\n", + "\n", + "In order to retain close that gap, the `measure` module now ships `MeasureConfig` objects, which handle the reading, writing and \"declarative\" defining of `Measure` objects.\n", + "\n", + "`Measure` objects can be instantiated from `MeasureConfig` objects using `Measure.from_config()`.\n", + "\n", + "### Summary of `Measure` vs `MeasureConfig`\n", + "\n", + "| `Measure` | `MeasureConfig` |\n", + "|-----------|--------------------|\n", + "| Is used for the actual computation | Is transformed into a `Measure` for actual computation |\n", + "| Uses python function to define what change to apply to the `Exposures`, `ImpactFuncSet`, `Hazard` objects | Define the changes (functions) to apply via the former way (scaling/shifting effect, alternate file loading, etc.) |\n", + "| Accepts any possible effect as long as it can be defined as a python function | Is restricted to a set of defined effects |\n", + "| Cannot be written to a file (unless it was created by a `MeasureConfig`) | Can easily be read from/written to a file (`.xlsx` or `.yaml`) |" + ] + }, + { + "cell_type": "markdown", + "id": "6d786faa-5b8c-4ee6-83cd-5fdafc1b2c29", + "metadata": {}, + "source": [ + "### Configuration classes\n", + "\n", + "The definition of measures via `MeasureConfig` is organized into a hierarchy of specialized classes:\n", + "\n", + "- `MeasureConfig`: The top-level container for a single measure.\n", + "- `HazardModifierConfig`: Defines how the hazard is changed (e.g., shifting intensity).\n", + "- `ImpfsetModifierConfig`: Adjusts impact functions (e.g., scaling vulnerability curves).\n", + "- `ExposuresModifierConfig`: Modifies exposure data (e.g., reassigning IDs or zeroing regions).\n", + "- `CostIncomeConfig`: Handles the financial aspects, including initial costs and recurring income.\n", + "\n", + "Note that everything can be defined and accessed directly from the `MeasureConfig` container, the underlying ones are there to keep things organized.\n", + "\n", + "In the following we present each of these subclasses and the possibilities they offer." + ] + }, + { + "cell_type": "markdown", + "id": "4887d2a6-8295-4fda-8442-cbcbd3b16fea", + "metadata": {}, + "source": [ + "## Quickstart" + ] + }, + { + "cell_type": "markdown", + "id": "5c40640d-50a4-4102-8e45-0dc8b9a770f2", + "metadata": {}, + "source": [ + "You can directly define a `MeasureConfig` object with a dictionary, using `MeasureConfig.from_dict()`.\n", + "\n", + "Below are the possible parameters:\n", + "\n", + "| Scope | Parameter | Type | Description |\n", + "| :--- | :--- | :--- | :--- |\n", + "| **Top-Level** | `name` (required) | `str` | Unique identifier for the measure. |\n", + "| | `haz_type` (required) | `str` | The hazard type this measure targets (e.g., \"TC\", \"FL\"). |\n", + "| | `implementation_duration` | `str` | Pandas offset alias (e.g., \"2Y\") for implementation time. |\n", + "| | `color_rgb` | `tuple` | RGB triple (0-1 range) for plotting and visualization. |\n", + "| **Hazard** | `haz_int_mult` | `float` | Multiplier for hazard intensity (default: 1.0). |\n", + "| | `haz_int_add` | `float` | Additive offset for hazard intensity (default: 0.0). |\n", + "| | `new_hazard_path` | | Path to an HDF5 file to replace the current hazard. |\n", + "| | `impact_rp_cutoff` | `float` | Return period (years) threshold; events below this are ignored. |\n", + "| **Impact Function**| `impf_ids` | `list` | Specific impact function IDs to modify (None = all). |\n", + "| | `impf_mdd_mult` / `_add` | `float` | Scale or shift the Mean Damage Degree curve. |\n", + "| | `impf_paa_mult` / `_add` | `float` | Scale or shift the Percentage of Assets Affected curve. |\n", + "| | `impf_int_mult` / `_add` | `float` | Scale or shift the intensity axis of the function. |\n", + "| | `new_impfset_path` | | Path to an Excel file to replace the impact function set. |\n", + "| **Exposures** | `reassign_impf_id` | `dict` | Mapping `{haz_type: {old_id: new_id}}` for reclassification. |\n", + "| | `set_to_zero` | `list` | List of Region IDs where exposure value is set to 0. |\n", + "| | `new_exposures_path` | | Path to an HDF5 file to replace the current exposures. |\n", + "| **Cost & Income** | `init_cost` | `float` | One-time investment cost (absolute value). |\n", + "| | `periodic_cost` | `float` | Recurring maintenance/operational costs. |\n", + "| | `periodic_income` | `float` | Recurring income generated by the measure. |\n", + "| | `mkt_price_year` | `int` | Reference year for pricing (default: current year). |\n", + "| | `freq` | `str` | Frequency of cash flows (e.g., \"Y\" for yearly). |\n", + "| | `custom_cash_flows` | `list[dict]`| Explicit list of dates and values for complex cash flows. (See the [cost income tutorial](cost-income-tutorial)) |" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "62cf6502-7765-452c-be32-eb49a363b4a8", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "MeasureConfig(\n", + "\tname='Tutorial measure'\n", + "\thaz_type='TC'\n", + "\timpfset_modifier=ImpfsetModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\timpf_ids=[1, 2]\n", + "\t\t\timpf_mdd_mult=0.8\n", + ")\n", + "\thazard_modifier=HazardModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tnew_hazard_path='path/to/new_hazard.h5'\n", + ")\n", + "\texposures_modifier=ExposuresModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\treassign_impf_id={'TC': {1: 2}}\n", + ")\n", + "\tcost_income=CostIncomeConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tinit_cost=10000\n", + "\t\t\tperiodic_cost=500\n", + ")\n", + "\timplementation_duration=None\n", + "\tcolor_rgb=(0.1, 0.5, 0.3))\n" + ] + } + ], + "source": [ + "from climada.entity.measures.measure_config import MeasureConfig\n", + "\n", + "measure_dict = {\n", + " \"name\": \"Tutorial measure\",\n", + " \"haz_type\": \"TC\",\n", + " \"impf_ids\": [1, 2],\n", + " \"impf_mdd_mult\": 0.8,\n", + " \"new_hazard_path\": \"path/to/new_hazard.h5\",\n", + " \"reassign_impf_id\": {\"TC\": {1: 2}},\n", + " \"color_rgb\": [0.1, 0.5, 0.3],\n", + " \"init_cost\": 10000,\n", + " \"periodic_cost\": 500,\n", + "}\n", + "\n", + "meas_config = MeasureConfig.from_dict(measure_dict)\n", + "\n", + "print(meas_config)" + ] + }, + { + "cell_type": "markdown", + "id": "ac98393f-575f-4580-ac4a-dae578638916", + "metadata": {}, + "source": [ + "## Modifying Impact Functions: `ImpfsetModifierConfig`\n", + "\n", + "The `ImpfsetModifierConfig` is used to define how an adaptation measure changes the vulnerability (refer to the [impact functions tutorial](impact-functions-tutorial)).\n", + "\n", + "When \"translated\" to a `Measure` object the `ImpfsetModifierConfig` populates the `impfset_change` attribute with a function that takes an `ImpactFuncSet` and returns a modified one, according to the specifications.\n", + "\n", + "```{note}\n", + "Modifications are always applied to a specific hazard type (`haz_type` parameter).\n", + "```\n", + "\n", + "`ImpfsetModifierConfig` allows you to modify the main components of an impact function set, as well as to replace it entirely:\n", + "\n", + "- The MDD (Mean Damage Degree) array: via `impf_mdd_mult` to scale it and `impf_mdd_add` to shift it.\n", + "- The PAA (Percentage of Assets Affected) array: via `impf_paa_mult` to scale it and `impf_paa_add` to shift it.\n", + "- The intensity array: via `impf_int_mult` to scale it and `impf_int_add` to shift it.\n", + "- Replacing the set: via providing the `new_impfset_path` parameter. It needs to be a valid `.xlsx` file readable by `ImpactFuncSet.from_excel()`\n", + "\n", + "See below for code examples.\n", + "\n", + "```{warning}\n", + "If you provide a new_impfset_path and other modifiers, CLIMADA will load the new file first and then apply the modifiers to it. (A warning will be issued to ensure this sequence is intended).\n", + "```\n", + "\n", + "```{note}\n", + "By default the changes are applied to all the impact functions in the set, but you can provide the `impf_ids` parameter to apply the changes to a selection of impact function ids.\n", + "```" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "id": "5ffb447b-1b8f-4e40-9d1c-7db33a11255e", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- Scaling Config ---\n", + "ImpfsetModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\timpf_ids=[1, 2]\n", + "\t\t\timpf_mdd_mult=0.8\n", + "\t\t\timpf_int_add=5.0\n", + ")\n", + "\n", + "--- Replacement Config ---\n", + "ImpfsetModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tnew_impfset_path='path/to/new_impact_functions.xlsx'\n", + ")\n" + ] + } + ], + "source": [ + "from climada.entity.measures.measure_config import ImpfsetModifierConfig\n", + "\n", + "# 1. Scaling existing Impact Functions\n", + "# Let's say we want to simulate a 20% reduction in MDD\n", + "# and a slight shift in the intensity threshold for Hazard 'TC'.\n", + "impf_mod_scaling = ImpfsetModifierConfig(\n", + " haz_type=\"TC\",\n", + " impf_ids=[1, 2], # Apply only to specific function IDs\n", + " impf_mdd_mult=0.8, # Reduce Mean Damage Degree by 20%\n", + " impf_int_add=5.0, # Shift intensity axis by 5 units (e.g., higher resistance)\n", + ")\n", + "\n", + "print(\"--- Scaling Config ---\")\n", + "print(impf_mod_scaling)\n", + "\n", + "# 2. Replacing the Impact Function Set from a file\n", + "# Useful for measures that implement completely new building standards.\n", + "impf_mod_replace = ImpfsetModifierConfig(\n", + " haz_type=\"TC\", new_impfset_path=\"path/to/new_impact_functions.xlsx\"\n", + ")\n", + "\n", + "print(\"\\n--- Replacement Config ---\")\n", + "print(impf_mod_replace)" + ] + }, + { + "cell_type": "markdown", + "id": "234ebc89-83b0-42b1-8b97-734016306b84", + "metadata": {}, + "source": [ + "## Modifying Hazards: `HazardModifierConfig`\n", + "\n", + "The `HazardModifierConfig` is used to define how an adaptation measure changes the hazard (refer to the [hazard tutorial](hazard-tutorial)).\n", + "\n", + "When \"translated\" to a `Measure` object the `HazardModifierConfig` populates the `hazard_change` attribute with a function that takes a `Hazard` (possibly additional arguments, see below) and returns a modified one, according to the specifications.\n", + "\n", + "```{note}\n", + "Modifications are always applied to a specific hazard type (`haz_type` parameter).\n", + "```\n", + "\n", + "`HazardModifierConfig` allows you to modify the intensity and frequency of the hazard, to apply a cutoff on the return period of impacts, as well as to replace it entirely:\n", + "\n", + "- The intensity matrix: via `haz_int_mult` to scale it and `haz_int_add` to shift it.\n", + "- The frequency array: via `haz_freq_mult` to scale it and `haz_freq_add` to shift it.\n", + "- Replacing the hazard: via providing the `new_hazard_path` parameter. It needs to be a valid hazard HDF5 file readable by `Hazard.from_hdf5()`\n", + "- Applying a cutoff on frequency based on impacts: via `impact_rp_cutoff` (see the note).\n", + "\n", + "```{note}\n", + "Providing a value for `impact_rp_cutoff` \"removes\" (it sets their intensity to 0.) events from the hazard, for which the exceedance frequency (inverse of return period) of impacts is below the given threshold.\n", + "\n", + "For instance providing 1/20, would remove all events whose impacts have a return period below 20 years.\n", + "\n", + "In that case the function changing the hazard (`Measure.hazard_change`) will be a function with the following signature:\n", + "\n", + " f(hazard: Hazard, # The hazard to apply on\n", + " exposures: Exposures, # The exposure for the impact computation\n", + " impfset: ImpactFuncSet, # The impfset for the impact computation\n", + " base_hazard: Hazard, # The hazard for the impact computation\n", + " exposures_region_id: Optional[list[int]] = None, # Region id to filter to\n", + " ) -> Hazard\n", + "```\n", + "\n", + "```{warning}\n", + "If you provide a new_hazard_path and other modifiers, CLIMADA will load the new file first and then apply the modifiers to it. (A warning will be issued to ensure this sequence is intended).\n", + "```" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "id": "f6061c1c-b21f-4aef-a394-c172784a25ab", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- Scaling Config ---\n", + "HazardModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\thaz_int_add=-10\n", + "\t\t\thaz_freq_mult=0.8\n", + ")\n", + "\n", + "--- Replacement Config ---\n", + "HazardModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tnew_hazard_path='path/to/new_floods.h5'\n", + ")\n", + "\n", + "--- Cutoff Config ---\n", + "HazardModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\timpact_rp_cutoff=0.05\n", + ")\n" + ] + } + ], + "source": [ + "from climada.entity.measures.measure_config import HazardModifierConfig\n", + "\n", + "# 1. Scaling existing hazard\n", + "# Let's say we want to simulate a 20% reduction in frequency\n", + "# and a reduction by 10m/s in the intensity for our tropical cyclones.\n", + "haz_mod = HazardModifierConfig(\n", + " haz_type=\"TC\",\n", + " haz_int_add=-10, # Reduce hazard intensity by 10 units\n", + " haz_freq_mult=0.8, # Scale hazard frequency by 20%\n", + ")\n", + "\n", + "print(\"--- Scaling Config ---\")\n", + "print(haz_mod)\n", + "\n", + "# 2. Replacing the hazard from a file\n", + "# Useful for measures that correspond to a different hazard modelling.\n", + "# E.g., a dike leading to a change in (physical) flood modelling.\n", + "haz_mod_new = HazardModifierConfig(\n", + " haz_type=\"FL\", new_hazard_path=\"path/to/new_floods.h5\"\n", + ")\n", + "\n", + "print(\"\\n--- Replacement Config ---\")\n", + "print(haz_mod_new)\n", + "\n", + "# 3. Applying a cutoff on the return period of the impacts\n", + "# Useful when measures are defined to avoid damage for a specific RP (exceedance frequency).\n", + "# Note that it looks a the distribution of the impacts, not the hazard intensity!\n", + "haz_mod_cutoff = HazardModifierConfig(\n", + " haz_type=\"TC\",\n", + " impact_rp_cutoff=1\n", + " / 20, # Set intensity to 0 for events with impacts with a return period below 20 years\n", + ")\n", + "\n", + "print(\"\\n--- Cutoff Config ---\")\n", + "print(haz_mod_cutoff)" + ] + }, + { + "cell_type": "markdown", + "id": "c7499c1a-2491-42c4-bdbb-d224090b85fb", + "metadata": {}, + "source": [ + "## Modifying Exposures: `ExposuresModifierConfig`\n", + "\n", + "The `ExposuresModifierConfig` is used to define how an adaptation measure changes the exposure (refer to the [exposure tutorial](exposure-tutorial)).\n", + "\n", + "When \"translated\" to a `Measure` object the `ExposuresModifierConfig` populates the `exposures_change` attribute with a function that takes an `Exposures` and returns a modified one, according to the specifications.\n", + "\n", + "`ExposuresModifierConfig` allows you to modify the impact function assigned to different hazard, to set a list of points to 0 value, or to load a different Exposures:\n", + "\n", + "- Remapping the impact function: via `reassign_impf_id` with a dictionary of the form `{haz_type: {old_id: new_id}}`.\n", + "- Setting values to zero: via `set_to_zero` with a list of indices of the exposure GeoDataFrame.\n", + "- Replacing the exposure: via providing the `new_exposures_path` parameter. It need to be a valid HDF5 exposure file readable by `Exposures.from_hdf5()`\n", + "\n", + "```{warning}\n", + "If you provide a new_exposures_path and other modifiers, CLIMADA will load the new file first and then apply the modifiers to it. (A warning will be issued to ensure this sequence is intended).\n", + "```" + ] + }, + { + "cell_type": "code", + "execution_count": 4, + "id": "5237930d-a18c-4498-afe5-373c5dadf882", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- First Config ---\n", + "ExposuresModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\treassign_impf_id={'TC': {1: 2}}\n", + "\t\t\tset_to_zero=[0, 25, 78]\n", + ")\n", + "\n", + "--- Replacement Config ---\n", + "ExposuresModifierConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tnew_exposures_path='path/to/exposures.h5'\n", + ")\n" + ] + } + ], + "source": [ + "from climada.entity.measures.measure_config import ExposuresModifierConfig\n", + "\n", + "# 1. Changing existing Exposures\n", + "exp_mod = ExposuresModifierConfig(\n", + " reassign_impf_id={\"TC\": {1: 2}}, # Remaps exposures points with impf_TC == 1 to 2.\n", + " set_to_zero=[\n", + " 0,\n", + " 25,\n", + " 78,\n", + " ], # Sets the value of exposure points with index 0, 25 and 78 to 0.\n", + ")\n", + "\n", + "print(\"--- First Config ---\")\n", + "print(exp_mod)\n", + "\n", + "# 2. Replacing the expoosure from a file\n", + "exp_mod_new = ExposuresModifierConfig(new_exposures_path=\"path/to/exposures.h5\")\n", + "\n", + "print(\"\\n--- Replacement Config ---\")\n", + "print(exp_mod_new)" + ] + }, + { + "cell_type": "markdown", + "id": "2c2d4488-28e5-4ced-b9cf-e4d5c0cade3e", + "metadata": {}, + "source": [ + "## Defining the financial aspects of the measure\n", + "\n", + "For in depth description of CostIncome objects, refer to the [related tutorial](cost-income-tutorial).\n", + "\n", + "```{note}\n", + "The default for mkt_price_year if not provided is the current year.\n", + "```\n", + "\n", + "You can easily define the CostIncome object to be associated with the measure using `CostIncomeConfig`:" + ] + }, + { + "cell_type": "code", + "execution_count": 5, + "id": "7c107fc5-606b-4904-8b8e-059f846c2e39", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "\n", + "--- Growth & Income Config ---\n", + "CostIncomeConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tinit_cost=500000.0\n", + "\t\t\tperiodic_cost=20000.0\n", + "\t\t\tperiodic_income=100000.0\n", + "\t\t\tcost_yearly_growth_rate=0.02\n", + "\t\t\tincome_yearly_growth_rate=0.03\n", + ")\n", + "\n", + "--- Custom Schedule Config ---\n", + "CostIncomeConfig(\n", + "\t\tNon default fields:\n", + "\t\t\tcustom_cash_flows=[{'date': '2024-01-01', 'value': -1000000}, {'date': '2029-01-01', 'value': -200000}, {'date': '2034-01-01', 'value': 500000}]\n", + ")\n" + ] + } + ], + "source": [ + "from climada.entity.measures.measure_config import CostIncomeConfig\n", + "\n", + "# This models a measure where costs increase by 2% annually,\n", + "# but it generates 100k in yearly income which grows by 3%.\n", + "growth_finance = CostIncomeConfig(\n", + " init_cost=500_000.0,\n", + " periodic_cost=20_000.0,\n", + " cost_yearly_growth_rate=0.02,\n", + " periodic_income=100_000.0,\n", + " income_yearly_growth_rate=0.03,\n", + " freq=\"Y\",\n", + ")\n", + "\n", + "print(\"\\n--- Growth & Income Config ---\")\n", + "print(growth_finance)\n", + "\n", + "\n", + "# Custom Cash Flow\n", + "# If the investment isn't linear (e.g., a major retrofit in year 5),\n", + "# you can define a list of specific events.\n", + "custom_schedule = [\n", + " {\"date\": \"2024-01-01\", \"value\": -1000000}, # Initial cost\n", + " {\"date\": \"2029-01-01\", \"value\": -200000}, # Mid-term overhaul\n", + " {\"date\": \"2034-01-01\", \"value\": 500000}, # Terminal value\n", + "]\n", + "\n", + "custom_finance = CostIncomeConfig(custom_cash_flows=custom_schedule)\n", + "\n", + "print(\"\\n--- Custom Schedule Config ---\")\n", + "print(custom_finance)" + ] + }, + { + "cell_type": "markdown", + "id": "ab4216dd-fd0b-4939-844d-56bd5ea49504", + "metadata": {}, + "source": [ + "## Reading from and writing to\n", + "\n", + "You can easily write/read measure configurations from YAML, as well as from pandas Series.\n", + "\n", + "You can also create `Measures`/`MeasureSet` directly, using the same methods (these methods first load the file as a `MeasureConfig` and convert it directly to a `Measure`)\n", + "Similarly you can still create `MeasureSet` from legacy Excel or matlab files using `MeasureSet.from_excel()` which takes care of remapping the legacy parameter names to the new ones.\n", + "See the [measure tutorial](measure-tutorial) for more details on that." + ] + }, + { + "cell_type": "markdown", + "id": "63132690-dd6f-4f45-96a9-5519fa2dec07", + "metadata": {}, + "source": [ + "\n", + "```python\n", + "import pandas as pd\n", + "from climada.entity.measures.measure_config import MeasureConfig\n", + "\n", + "# 1. Exporting to YAML\n", + "# Assuming 'my_measure_config' is a MeasureConfig object created previously\n", + "my_measure_config.to_yaml(\"seawall_config.yaml\")\n", + "\n", + "# 2. Loading from YAML\n", + "loaded_measure_config = MeasureConfig.from_yaml(\"seawall_config.yaml\")\n", + "\n", + "# 3. Loading from Pandas\n", + "row_data = pd.Series({\n", + " \"name\": \"Mangrove_Restoration\",\n", + " \"haz_type\": \"TC\",\n", + " \"impf_mdd_mult\": 0.7,\n", + " \"init_cost\": 250000,\n", + " \"color_rgb\": (0.1, 0.8, 0.1)\n", + "})\n", + "\n", + "pandas_measure_config = MeasureConfig.from_row(row_data)\n", + "\n", + "# 4. Measure object directly\n", + "measure = Measure.from_yaml(\"seawall_config.yaml\")\n", + "```" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python [conda env:climada_env_dev]", + "language": "python", + "name": "conda-env-climada_env_dev-py" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.15" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}