Risk Trajectory Split 1 : Snapshots

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 <https://www.gnu.org/licenses/>.
+
+---
+
+This module implements risk trajectory objects which enable computation and
+possibly interpolation of risk metric over multiple dates.
+
+"""
+
+from .snapshot import Snapshot
+
+__all__ = [
+    "Snapshot",
+]

climada/trajectories/snapshot.py

@@ -0,0 +1,167 @@
+"""
+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 <https://www.gnu.org/licenses/>.
+
+---
+
+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
+
+import pandas as pd
+
+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,
+        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 = None
+        self._date = self._convert_to_date(date)
+
+    @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)
+        elif 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:
+                raise ValueError("String must be in the format 'YYYY-MM-DD'")
+        elif isinstance(date_arg, datetime.date):
+            # Already a date object
+            return date_arg
+        else:
+            raise TypeError("date_arg must be an int, str, or datetime.date")
+
+    def apply_measure(self, measure: Measure) -> "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(f"Applying measure {measure.name} on snapshot {id(self)}")
+        exp, impfset, haz = measure.apply(self.exposure, self.impfset, self.hazard)
+        snap = Snapshot(exposure=exp, hazard=haz, impfset=impfset, date=self.date)
+        snap._measure = measure
+        return snap

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)