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,202 @@
+"""
+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
+from typing import cast
+
+import numpy as np
+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 : datetime.date | str | pd.Timestamp
+        The date of the Snapshot, it can be an string representing a year,
+        a datetime object or a string representation of a datetime object.
+    measure : Measure | None, default None.
+        Measure associated with the Snapshot. The measure object is *not* applied
+        to the other parameters of the object (Exposure, Hazard, Impfset).
+        To create a `Snapshot` with a measure use `apply_measure()` instead (see notes).
+        The use of anything but None should be reserved for advanced users.
+    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
+        A possible measure associated with the snapshot.
+
+    Notes
+    -----
+
+    Providing a measure to the init assumes that the (Exposure, Hazard, Impfset) triplet
+    already corresponds to the triplet once the measure is applied. Measure objects
+    contain "the changes to apply". Creating a consistent Snapshot with a measure should
+    be done by first creating a Snapshot with the "baseline" (Exposure, Hazard, Impfset) triplet
+    and calling `<Snapshot>.apply_measure(<measure>)`, which returns a new Snapshot object
+    with the measure applied.
+
+    Instantiating a Snapshot with a measure directly does not garantee the
+    consistency between the triplet and the measure, and should be avoided.
+
+    If `ref_only` is True (default) 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 snapshots as immutable objects.
+
+    """
+
+    def __init__(
+        self,
+        *,
+        exposure: Exposures,
+        hazard: Hazard,
+        impfset: ImpactFuncSet,
+        date: datetime.date | str | pd.Timestamp,
+        measure: Measure | None = None,
+        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(measure)
+        self._date = self._convert_to_timestamp(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) -> pd.Timestamp:
+        """Date of the snapshot."""
+        return self._date
+
+    @property
+    def impact_calc_kwargs(self) -> dict:
+        """Convenience function for ImpactCalc class."""
+        return {
+            "exposures": self.exposure,
+            "hazard": self.hazard,
+            "impfset": self.impfset,
+        }
+
+    @staticmethod
+    def _convert_to_timestamp(
+        date_arg: str | datetime.date | pd.Timestamp | np.datetime64,
+    ) -> pd.Timestamp:
+        """
+        Convert date argument of type str, datetime.date,
+        np.datetime64, or pandas Timestamp to a pandas Timestamp object.
+        """
+        if isinstance(date_arg, str):
+            try:
+                date = pd.Timestamp(date_arg)
+            except (ValueError, TypeError) as exc:
+                raise ValueError(
+                    "String must be in a valid date format (e.g., 'YYYY-MM-DD')"
+                ) from exc
+
+        elif isinstance(date_arg, (datetime.date, pd.Timestamp, np.datetime64)):
+            date = pd.Timestamp(date_arg)
+
+        else:
+            raise TypeError(
+                f"Unsupported type: {type(date_arg)}. Must be str, date, Timestamp, or datetime64."
+            )
+
+        # Final check for NaT (Not-a-Time)
+        if date is pd.NaT:
+            raise ValueError(
+                f"Could not resolve '{date_arg}' to a valid Pandas Timestamp."
+            )
+
+        return cast(pd.Timestamp, 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("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=True,  # Avoid unecessary copies of new objects
+        )
+        return snap

climada/trajectories/test/test_snapshot.py

@@ -0,0 +1,162 @@
+import datetime
+from unittest.mock import MagicMock
+
+import numpy as np
+import pandas as pd
+import pytest
+
+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
+
+# --- Fixtures ---
+
+
+@pytest.fixture(scope="module")
+def shared_data():
+    """Load heavy HDF5 data once per module to speed up tests."""
+    exposure = Exposures.from_hdf5(EXP_DEMO_H5)
+    hazard = Hazard.from_hdf5(HAZ_DEMO_H5)
+    impfset = ImpactFuncSet(
+        [
+            ImpactFunc(
+                "TC",
+                3,
+                intensity=np.array([0, 20]),
+                mdd=np.array([0, 0.5]),
+                paa=np.array([0, 1]),
+            )
+        ]
+    )
+    return exposure, hazard, impfset
+
+
+@pytest.fixture
+def mock_context(shared_data):
+    """Provides the exposure/hazard/impfset and a pre-configured mock measure."""
+    exp, haz, impf = shared_data
+
+    # Setup Mock Measure
+    mock_measure = MagicMock(spec=Measure)
+    mock_measure.name = "Test Measure"
+
+    modified_exp = MagicMock(spec=Exposures)
+    modified_haz = MagicMock(spec=Hazard)
+    modified_imp = MagicMock(spec=ImpactFuncSet)
+
+    mock_measure.apply.return_value = (modified_exp, modified_imp, modified_haz)
+
+    return {
+        "exp": exp,
+        "haz": haz,
+        "imp": impf,
+        "measure": mock_measure,
+        "mod_exp": modified_exp,
+        "mod_haz": modified_haz,
+        "mod_imp": modified_imp,
+        "date": pd.Timestamp("2023"),
+    }
+
+
+# --- Tests ---
+
+
+@pytest.mark.parametrize(
+    "input_date,expected",
+    [
+        ("2023", pd.Timestamp(2023, 1, 1)),
+        ("2023-01-01", pd.Timestamp(2023, 1, 1)),
+        (np.datetime64("2023-01-01"), pd.Timestamp(2023, 1, 1)),
+        (datetime.date(2023, 1, 1), pd.Timestamp(2023, 1, 1)),
+        (pd.Timestamp(2023, 1, 1), pd.Timestamp(2023, 1, 1)),
+    ],
+)
+def test_init_valid_dates(mock_context, input_date, expected):
+    """Test various valid date input formats using parametrization."""
+    snapshot = Snapshot(
+        exposure=mock_context["exp"],
+        hazard=mock_context["haz"],
+        impfset=mock_context["imp"],
+        date=input_date,
+    )
+    assert snapshot.date == expected
+
+
+def test_init_invalid_date_format(mock_context):
+    with pytest.raises(ValueError, match=r"String must be in a valid date format"):
+        Snapshot(
+            exposure=mock_context["exp"],
+            hazard=mock_context["haz"],
+            impfset=mock_context["imp"],
+            date="invalid-date",
+        )
+
+
+def test_init_invalid_date_type(mock_context):
+    with pytest.raises(
+        TypeError,
+        match=r"Unsupported type",
+    ):
+        Snapshot(
+            exposure=mock_context["exp"],
+            hazard=mock_context["haz"],
+            impfset=mock_context["imp"],
+            date=2023.5,  # type: ignore
+        )
+
+
+def test_properties(mock_context):
+    snapshot = Snapshot(
+        exposure=mock_context["exp"],
+        hazard=mock_context["haz"],
+        impfset=mock_context["imp"],
+        date=mock_context["date"],
+    )
+
+    # Check that it's a deep copy (new reference)
+    assert snapshot.exposure is not mock_context["exp"]
+    assert snapshot.hazard is not mock_context["haz"]
+
+    assert snapshot.measure is None
+
+    # Check data equality
+    pd.testing.assert_frame_equal(snapshot.exposure.gdf, mock_context["exp"].gdf)
+    assert snapshot.hazard.haz_type == mock_context["haz"].haz_type
+    assert snapshot.impfset == mock_context["imp"]
+    assert snapshot.date == mock_context["date"]
+
+
+def test_reference(mock_context):
+    snapshot = Snapshot(
+        exposure=mock_context["exp"],
+        hazard=mock_context["haz"],
+        impfset=mock_context["imp"],
+        date=mock_context["date"],
+        ref_only=True,
+    )
+
+    # Check that it is a reference
+    assert snapshot.exposure is mock_context["exp"]
+    assert snapshot.hazard is mock_context["haz"]
+    assert snapshot.impfset is mock_context["imp"]
+    assert snapshot.measure is None
+
+
+def test_apply_measure(mock_context):
+    snapshot = Snapshot(
+        exposure=mock_context["exp"],
+        hazard=mock_context["haz"],
+        impfset=mock_context["imp"],
+        date=mock_context["date"],
+    )
+    new_snapshot = snapshot.apply_measure(mock_context["measure"])
+
+    assert new_snapshot.measure is not None
+    assert new_snapshot.measure.name == "Test Measure"
+    assert new_snapshot.exposure == mock_context["mod_exp"]
+    assert new_snapshot.hazard == mock_context["mod_haz"]
+    assert new_snapshot.impfset == mock_context["mod_imp"]
+    assert new_snapshot.date == mock_context["date"]