diff --git a/climada/trajectories/__init__.py b/climada/trajectories/__init__.py new file mode 100644 index 0000000000..91aca62d1c --- /dev/null +++ b/climada/trajectories/__init__.py @@ -0,0 +1,28 @@ +""" +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 risk trajectory objects which enable computation and +possibly interpolation of risk metric over multiple dates. + +""" + +from .snapshot import Snapshot + +__all__ = [ + "Snapshot", +] diff --git a/climada/trajectories/impact_calc_strat.py b/climada/trajectories/impact_calc_strat.py new file mode 100644 index 0000000000..b1bb6eebd3 --- /dev/null +++ b/climada/trajectories/impact_calc_strat.py @@ -0,0 +1,114 @@ +""" +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 modules implements the impact computation strategy objects for risk +trajectories. + +""" + +from abc import ABC, abstractmethod + +from climada.engine.impact import Impact +from climada.engine.impact_calc import ImpactCalc +from climada.entity.exposures.base import Exposures +from climada.entity.impact_funcs.impact_func_set import ImpactFuncSet +from climada.hazard.base import Hazard + +__all__ = ["ImpactCalcComputation"] + + +# The following is acceptable. +# We design a pattern, and currently it requires only to +# define the compute_impacts method. +# pylint: disable=too-few-public-methods +class ImpactComputationStrategy(ABC): + """ + Interface for impact computation strategies. + + This abstract class defines the contract for all concrete strategies + responsible for calculating and optionally modifying with a risk transfer, + the impact computation, based on a set of inputs (exposure, hazard, vulnerability). + + It revolves around a `compute_impacts()` method that takes as arguments + the three dimensions of risk (exposure, hazard, vulnerability) and return an + Impact object. + """ + + @abstractmethod + def compute_impacts( + self, + exp: Exposures, + haz: Hazard, + vul: ImpactFuncSet, + ) -> Impact: + """ + Calculates the total impact, including optional risk transfer application. + + Parameters + ---------- + exp : Exposures + The exposure data. + haz : Hazard + The hazard data (e.g., event intensity). + vul : ImpactFuncSet + The set of vulnerability functions. + + Returns + ------- + Impact + An object containing the computed total impact matrix and metrics. + + See Also + -------- + ImpactCalcComputation : The default implementation of this interface. + """ + + +class ImpactCalcComputation(ImpactComputationStrategy): + r""" + Default impact computation strategy using the core engine of climada. + + This strategy first calculates the raw impact using the standard + :class:`ImpactCalc` logic. + + """ + + def compute_impacts( + self, + exp: Exposures, + haz: Hazard, + vul: ImpactFuncSet, + ) -> Impact: + """ + Calculates the impact. + + Parameters + ---------- + exp : Exposures + The exposure data. + haz : Hazard + The hazard data. + vul : ImpactFuncSet + The set of vulnerability functions. + + Returns + ------- + Impact + The final impact object. + """ + return ImpactCalc(exposures=exp, impfset=vul, hazard=haz).impact() diff --git a/climada/trajectories/snapshot.py b/climada/trajectories/snapshot.py new file mode 100644 index 0000000000..8d9a74ef6f --- /dev/null +++ b/climada/trajectories/snapshot.py @@ -0,0 +1,221 @@ +""" +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 modules implements the Snapshot class. + +Snapshot are used to store a snapshot of Exposure, Hazard and Vulnerability +at a specific date. + +""" + +import copy +import datetime +import logging + +from climada.entity.exposures import Exposures +from climada.entity.impact_funcs import ImpactFuncSet +from climada.entity.measures.base import Measure +from climada.hazard import Hazard + +LOGGER = logging.getLogger(__name__) + +__all__ = ["Snapshot"] + + +class Snapshot: + """ + A snapshot of exposure, hazard, and impact function at a specific date. + + Parameters + ---------- + exposure : Exposures + hazard : Hazard + impfset : ImpactFuncSet + date : int | datetime.date | str + The date of the Snapshot, it can be an integer representing a year, + a datetime object or a string representation of a datetime object + with format "YYYY-MM-DD". + ref_only : bool, default False + Should the `Snapshot` contain deep copies of the Exposures, Hazard and Impfset (False) + or references only (True). + + Attributes + ---------- + date : datetime + Date of the snapshot. + measure: Measure | None + The possible measure applied to the snapshot. + + Notes + ----- + + The object creates deep copies of the exposure hazard and impact function set. + + Also note that exposure, hazard and impfset are read-only properties. + Consider snapshot as immutable objects. + + To create a snapshot with a measure, create a snapshot `snap` without + the measure and call `snap.apply_measure(measure)`, which returns a new Snapshot object + with the measure applied to its risk dimensions. + """ + + def __init__( + self, + *, + exposure: Exposures, + hazard: Hazard, + impfset: ImpactFuncSet, + measure: Measure | None, + date: int | datetime.date | str, + ref_only: bool = False, + ) -> None: + self._exposure = exposure if ref_only else copy.deepcopy(exposure) + self._hazard = hazard if ref_only else copy.deepcopy(hazard) + self._impfset = impfset if ref_only else copy.deepcopy(impfset) + self._measure = measure if ref_only else copy.deepcopy(impfset) + self._date = self._convert_to_date(date) + + @classmethod + def from_triplet( + cls, + *, + exposure: Exposures, + hazard: Hazard, + impfset: ImpactFuncSet, + date: int | datetime.date | str, + ref_only: bool = False, + ) -> "Snapshot": + """Create a Snapshot from exposure, hazard and impact functions set + + This method is the main point of entry for the creation of Snapshot. It + creates a new Snapshot object for the given date with copies of the + hazard, exposure and impact function set given in argument (or + references if ref_only is True) + + Parameters + ---------- + exposure : Exposures + hazard : Hazard + impfset : ImpactFuncSet + date : int | datetime.date | str + ref_only : bool + If true, uses references to the exposure, hazard and impact + function objects. Note that modifying the original objects after + computations using the Snapshot might lead to inconsistencies in + results. + + Returns + ------- + Snapshot + + Notes + ----- + + To create a Snapshot with a measure, first create the Snapshot without + the measure using this method, and use `apply_measure(measure)` afterward. + + """ + return cls( + exposure=exposure, + hazard=hazard, + impfset=impfset, + measure=None, + date=date, + ref_only=ref_only, + ) + + @property + def exposure(self) -> Exposures: + """Exposure data for the snapshot.""" + return self._exposure + + @property + def hazard(self) -> Hazard: + """Hazard data for the snapshot.""" + return self._hazard + + @property + def impfset(self) -> ImpactFuncSet: + """Impact function set data for the snapshot.""" + return self._impfset + + @property + def measure(self) -> Measure | None: + """(Adaptation) Measure data for the snapshot.""" + return self._measure + + @property + def date(self) -> datetime.date: + """Date of the snapshot.""" + return self._date + + @property + def impact_calc_data(self) -> dict: + """Convenience function for ImpactCalc class.""" + return { + "exposures": self.exposure, + "hazard": self.hazard, + "impfset": self.impfset, + } + + @staticmethod + def _convert_to_date(date_arg) -> datetime.date: + """Convert date argument of type int or str to a datetime.date object.""" + if isinstance(date_arg, int): + # Assume the integer represents a year + return datetime.date(date_arg, 1, 1) + if isinstance(date_arg, str): + # Try to parse the string as a date + try: + return datetime.datetime.strptime(date_arg, "%Y-%m-%d").date() + except ValueError as exc: + raise ValueError("String must be in the format 'YYYY-MM-DD'") from exc + if isinstance(date_arg, datetime.date): + # Already a date object + return date_arg + + raise TypeError("date_arg must be an int, str, or datetime.date") + + def apply_measure(self, measure: Measure, ref_only: bool = False) -> "Snapshot": + """Create a new snapshot by applying a Measure object. + + This method creates a new `Snapshot` object by applying a measure on + the current one. + + Parameters + ---------- + measure : Measure + The measure to be applied to the snapshot. + + Returns + ------- + The Snapshot with the measure applied. + + """ + + LOGGER.debug("Applying measure %s on snapshot %s", measure.name, id(self)) + exp, impfset, haz = measure.apply(self.exposure, self.impfset, self.hazard) + snap = Snapshot( + exposure=exp, + hazard=haz, + impfset=impfset, + date=self.date, + measure=measure, + ref_only=ref_only, + ) + return snap diff --git a/climada/trajectories/test/test_impact_calc_strat.py b/climada/trajectories/test/test_impact_calc_strat.py new file mode 100644 index 0000000000..eb5a53a2c0 --- /dev/null +++ b/climada/trajectories/test/test_impact_calc_strat.py @@ -0,0 +1,97 @@ +""" +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 impact_calc_strat + +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from climada.engine import Impact +from climada.entity import ImpactFuncSet +from climada.entity.exposures import Exposures +from climada.hazard import Hazard +from climada.trajectories import Snapshot +from climada.trajectories.impact_calc_strat import ( + ImpactCalcComputation, + ImpactComputationStrategy, +) + +# --- Fixtures --- + + +@pytest.fixture +def mock_snapshot(): + """Provides a snapshot with mocked exposure, hazard, and impact functions.""" + snap = MagicMock(spec=Snapshot) + snap.exposure = MagicMock(spec=Exposures) + snap.hazard = MagicMock(spec=Hazard) + snap.impfset = MagicMock(spec=ImpactFuncSet) + return snap + + +@pytest.fixture +def strategy(): + """Provides an instance of the ImpactCalcComputation strategy.""" + return ImpactCalcComputation() + + +# --- Tests --- +def test_interface_compliance(strategy): + """Ensure the class correctly inherits from the Abstract Base Class.""" + assert isinstance(strategy, ImpactComputationStrategy) + assert isinstance(strategy, ImpactCalcComputation) + + +def test_compute_impacts(strategy, mock_snapshot): + """Test that compute_impacts calls the pre-transfer method correctly.""" + mock_impacts = MagicMock(spec=Impact) + + # We patch the ImpactCalc within trajectories + with patch("climada.trajectories.impact_calc_strat.ImpactCalc") as mock_ImpactCalc: + mock_ImpactCalc.return_value.impact.return_value = mock_impacts + result = strategy.compute_impacts( + exp=mock_snapshot.exposure, + haz=mock_snapshot.hazard, + vul=mock_snapshot.impfset, + ) + mock_ImpactCalc.assert_called_once_with( + exposures=mock_snapshot.exposure, + impfset=mock_snapshot.impfset, + hazard=mock_snapshot.hazard, + ) + mock_ImpactCalc.return_value.impact.assert_called_once() + assert result == mock_impacts + + +def test_cannot_instantiate_abstract_base_class(): + """Ensure ImpactComputationStrategy cannot be instantiated directly.""" + with pytest.raises(TypeError, match="Can't instantiate abstract class"): + ImpactComputationStrategy() # type: ignore + + +@pytest.mark.parametrize("invalid_input", [None, 123, "string"]) +def test_compute_impacts_type_errors(strategy, invalid_input): + """ + Smoke test: Ensure that if ImpactCalc raises errors due to bad input, + the strategy correctly propagates them. + """ + with pytest.raises(AttributeError): + strategy.compute_impacts(invalid_input, invalid_input, invalid_input) diff --git a/climada/trajectories/test/test_snapshot.py b/climada/trajectories/test/test_snapshot.py new file mode 100644 index 0000000000..4e3b465d8e --- /dev/null +++ b/climada/trajectories/test/test_snapshot.py @@ -0,0 +1,132 @@ +import datetime +import unittest +from unittest.mock import MagicMock + +import numpy as np +import pandas as pd + +from climada.entity.exposures import Exposures +from climada.entity.impact_funcs import ImpactFunc, ImpactFuncSet +from climada.entity.measures.base import Measure +from climada.hazard import Hazard +from climada.trajectories.snapshot import Snapshot +from climada.util.constants import EXP_DEMO_H5, HAZ_DEMO_H5 + + +class TestSnapshot(unittest.TestCase): + + def setUp(self): + # Create mock objects for testing + self.mock_exposure = Exposures.from_hdf5(EXP_DEMO_H5) + self.mock_hazard = Hazard.from_hdf5(HAZ_DEMO_H5) + self.mock_impfset = ImpactFuncSet( + [ + ImpactFunc( + "TC", + 3, + intensity=np.array([0, 20]), + mdd=np.array([0, 0.5]), + paa=np.array([0, 1]), + ) + ] + ) + self.mock_measure = MagicMock(spec=Measure) + self.mock_measure.name = "Test Measure" + + # Setup mock return values for measure.apply + self.mock_modified_exposure = MagicMock(spec=Exposures) + self.mock_modified_hazard = MagicMock(spec=Hazard) + self.mock_modified_impfset = MagicMock(spec=ImpactFuncSet) + self.mock_measure.apply.return_value = ( + self.mock_modified_exposure, + self.mock_modified_impfset, + self.mock_modified_hazard, + ) + + def test_init_with_int_date(self): + snapshot = Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date=2023, + ) + self.assertEqual(snapshot.date, datetime.date(2023, 1, 1)) + + def test_init_with_str_date(self): + snapshot = Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date="2023-01-01", + ) + self.assertEqual(snapshot.date, datetime.date(2023, 1, 1)) + + def test_init_with_date_object(self): + date_obj = datetime.date(2023, 1, 1) + snapshot = Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date=date_obj, + ) + self.assertEqual(snapshot.date, date_obj) + + def test_init_with_invalid_date(self): + with self.assertRaises(ValueError): + Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date="invalid-date", + ) + + def test_init_with_invalid_type(self): + with self.assertRaises(TypeError): + Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date=2023.5, # type: ignore + ) + + def test_properties(self): + snapshot = Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date=2023, + ) + + # We want a new reference + self.assertIsNot(snapshot.exposure, self.mock_exposure) + self.assertIsNot(snapshot.hazard, self.mock_hazard) + self.assertIsNot(snapshot.impfset, self.mock_impfset) + + # But we want equality + pd.testing.assert_frame_equal(snapshot.exposure.gdf, self.mock_exposure.gdf) + + self.assertEqual(snapshot.hazard.haz_type, self.mock_hazard.haz_type) + self.assertEqual(snapshot.hazard.intensity.nnz, self.mock_hazard.intensity.nnz) + self.assertEqual(snapshot.hazard.size, self.mock_hazard.size) + + self.assertEqual(snapshot.impfset, self.mock_impfset) + + def test_apply_measure(self): + snapshot = Snapshot( + exposure=self.mock_exposure, + hazard=self.mock_hazard, + impfset=self.mock_impfset, + date=2023, + ) + new_snapshot = snapshot.apply_measure(self.mock_measure) + + self.assertIsNotNone(new_snapshot.measure) + self.assertEqual(new_snapshot.measure.name, "Test Measure") # type: ignore + self.assertEqual(new_snapshot.exposure, self.mock_modified_exposure) + self.assertEqual(new_snapshot.hazard, self.mock_modified_hazard) + self.assertEqual(new_snapshot.impfset, self.mock_modified_impfset) + + +if __name__ == "__main__": + TESTS = unittest.TestLoader().loadTestsFromTestCase(TestSnapshot) + unittest.TextTestRunner(verbosity=2).run(TESTS) diff --git a/doc/api/climada/climada.rst b/doc/api/climada/climada.rst index 557532912f..2e8d053946 100644 --- a/doc/api/climada/climada.rst +++ b/doc/api/climada/climada.rst @@ -7,4 +7,5 @@ Software documentation per package climada.engine climada.entity climada.hazard + climada.trajectories climada.util diff --git a/doc/api/climada/climada.trajectories.impact_calc_strat.rst b/doc/api/climada/climada.trajectories.impact_calc_strat.rst new file mode 100644 index 0000000000..1bf211b4c0 --- /dev/null +++ b/doc/api/climada/climada.trajectories.impact_calc_strat.rst @@ -0,0 +1,7 @@ +climada\.trajectories\.impact_calc_strat module +---------------------------------------- + +.. automodule:: climada.trajectories.impact_calc_strat + :members: + :undoc-members: + :show-inheritance: diff --git a/doc/api/climada/climada.trajectories.rst b/doc/api/climada/climada.trajectories.rst new file mode 100644 index 0000000000..883078074f --- /dev/null +++ b/doc/api/climada/climada.trajectories.rst @@ -0,0 +1,8 @@ + +climada\.trajectories module +============================ + +.. toctree:: + + climada.trajectories.snapshot + climada.trajectories.impact_calc_strat diff --git a/doc/api/climada/climada.trajectories.snapshot.rst b/doc/api/climada/climada.trajectories.snapshot.rst new file mode 100644 index 0000000000..ba0faf57ac --- /dev/null +++ b/doc/api/climada/climada.trajectories.snapshot.rst @@ -0,0 +1,7 @@ +climada\.trajectories\.snapshot module +---------------------------------------- + +.. automodule:: climada.trajectories.snapshot + :members: + :undoc-members: + :show-inheritance: