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..cf93171e0e --- /dev/null +++ b/climada/trajectories/snapshot.py @@ -0,0 +1,232 @@ +""" +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 +import warnings + +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, + _from_factory: bool = False, + ) -> None: + if not _from_factory: + warnings.warn( + "Direct instantiation of 'Snapshot' is discouraged. " + "Use 'Snapshot.from_triplet()' instead.", + UserWarning, + stacklevel=2, + ) + 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_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, + _from_factory=True, + ) + + @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) -> "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 + _from_factory=True, + ) + return snap diff --git a/climada/trajectories/test/test_snapshot.py b/climada/trajectories/test/test_snapshot.py new file mode 100644 index 0000000000..cecfa39395 --- /dev/null +++ b/climada/trajectories/test/test_snapshot.py @@ -0,0 +1,168 @@ +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, + } + + +# --- 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, datetime.date(2023, 1, 1)), + ("2023-01-01", datetime.date(2023, 1, 1)), + (datetime.date(2023, 1, 1), datetime.date(2023, 1, 1)), + ], +) +def test_init_valid_dates(mock_context, input_date, expected): + """Test various valid date input formats using parametrization.""" + snapshot = Snapshot.from_triplet( + 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.from_triplet( + 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 int, str, or datetime.date" + ): + Snapshot.from_triplet(exposure=mock_context["exp"], hazard=mock_context["haz"], impfset=mock_context["imp"], date=2023.5) # type: ignore + + +def test_properties(mock_context): + snapshot = Snapshot.from_triplet( + exposure=mock_context["exp"], + hazard=mock_context["haz"], + impfset=mock_context["imp"], + date=2023, + ) + + # 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"] + + +def test_reference(mock_context): + snapshot = Snapshot.from_triplet( + exposure=mock_context["exp"], + hazard=mock_context["haz"], + impfset=mock_context["imp"], + date=2023, + 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 + + # 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"] + + +def test_apply_measure(mock_context): + snapshot = Snapshot.from_triplet( + exposure=mock_context["exp"], + hazard=mock_context["haz"], + impfset=mock_context["imp"], + date=2023, + ) + 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"] 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: