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/snapshot.py b/climada/trajectories/snapshot.py new file mode 100644 index 0000000000..277b2c0bdb --- /dev/null +++ b/climada/trajectories/snapshot.py @@ -0,0 +1,201 @@ +""" +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 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). + Users should leave it to None, and use `apply_measure()` instead (see notes). + 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. As Measure objects + contain "the changes to apply", the creating a consistent Snapshot with a measure should + be done by first creating a Snapshot with the "baseline" (Exposure, Hazard, Impfset) triplet + and calling `.apply_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 diff --git a/climada/trajectories/test/test_snapshot.py b/climada/trajectories/test/test_snapshot.py new file mode 100644 index 0000000000..63031c52fb --- /dev/null +++ b/climada/trajectories/test/test_snapshot.py @@ -0,0 +1,174 @@ +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 --- + + +def test_not_from_factory_warning(mock_context): + """Test that direct __init__ call raises a warning""" + with pytest.warns(UserWarning): + Snapshot( + exposure=mock_context["exp"], + hazard=mock_context["haz"], + impfset=mock_context["imp"], + measure=None, + date="2001", + ) + + +@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="String must be in the 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"date_arg must be an str, datetime.date or pandas.Timestamp", + ): + 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"] 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.rst b/doc/api/climada/climada.trajectories.rst new file mode 100644 index 0000000000..28c035e20e --- /dev/null +++ b/doc/api/climada/climada.trajectories.rst @@ -0,0 +1,7 @@ + +climada\.trajectories module +============================ + +.. toctree:: + + climada.trajectories.snapshot 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: