Skip to content

Commit 5efad83

Browse files
spjuhelspjuhel
committed
cherry pick snapshots from feature/risk_trajectories
1 parent f3c2f8d commit 5efad83

File tree

2 files changed

+295
-0
lines changed

2 files changed

+295
-0
lines changed

‎climada/trajectories/snapshot.py‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
"""
2+
This file is part of CLIMADA.
3+
4+
Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS.
5+
6+
CLIMADA is free software: you can redistribute it and/or modify it under the
7+
terms of the GNU General Public License as published by the Free
8+
Software Foundation, version 3.
9+
10+
CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY
11+
WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
12+
PARTICULAR PURPOSE. See the GNU General Public License for more details.
13+
14+
You should have received a copy of the GNU General Public License along
15+
with CLIMADA. If not, see <https://www.gnu.org/licenses/>.
16+
17+
---
18+
19+
This modules implements the Snapshot class.
20+
21+
Snapshot are used to store a snapshot of Exposure, Hazard and Vulnerability
22+
at a specific date.
23+
24+
"""
25+
26+
import copy
27+
import datetime
28+
import logging
29+
30+
import pandas as pd
31+
32+
from climada.entity.exposures import Exposures
33+
from climada.entity.impact_funcs import ImpactFuncSet
34+
from climada.entity.measures.base import Measure
35+
from climada.hazard import Hazard
36+
37+
LOGGER = logging.getLogger(__name__)
38+
39+
__all__ = ["Snapshot"]
40+
41+
42+
class Snapshot:
43+
"""
44+
A snapshot of exposure, hazard, and impact function at a specific date.
45+
46+
Parameters
47+
----------
48+
exposure : Exposures
49+
hazard : Hazard
50+
impfset : ImpactFuncSet
51+
date : int | datetime.date | str
52+
The date of the Snapshot, it can be an integer representing a year,
53+
a datetime object or a string representation of a datetime object
54+
with format "YYYY-MM-DD".
55+
56+
Attributes
57+
----------
58+
date : datetime
59+
Date of the snapshot.
60+
measure: Measure | None
61+
The possible measure applied to the snapshot.
62+
63+
Notes
64+
-----
65+
66+
The object creates deep copies of the exposure hazard and impact function set.
67+
68+
Also note that exposure, hazard and impfset are read-only properties.
69+
Consider snapshot as immutable objects.
70+
71+
To create a snapshot with a measure, create a snapshot `snap` without
72+
the measure and call `snap.apply_measure(measure)`, which returns a new Snapshot object
73+
with the measure applied to its risk dimensions.
74+
"""
75+
76+
def __init__(
77+
self,
78+
*,
79+
exposure: Exposures,
80+
hazard: Hazard,
81+
impfset: ImpactFuncSet,
82+
date: int | datetime.date | str,
83+
) -> None:
84+
self._exposure = copy.deepcopy(exposure)
85+
self._hazard = copy.deepcopy(hazard)
86+
self._impfset = copy.deepcopy(impfset)
87+
self._measure = None
88+
self._date = self._convert_to_date(date)
89+
90+
@property
91+
def exposure(self) -> Exposures:
92+
"""Exposure data for the snapshot."""
93+
return self._exposure
94+
95+
@property
96+
def hazard(self) -> Hazard:
97+
"""Hazard data for the snapshot."""
98+
return self._hazard
99+
100+
@property
101+
def impfset(self) -> ImpactFuncSet:
102+
"""Impact function set data for the snapshot."""
103+
return self._impfset
104+
105+
@property
106+
def measure(self) -> Measure | None:
107+
"""(Adaptation) Measure data for the snapshot."""
108+
return self._measure
109+
110+
@property
111+
def date(self) -> datetime.date:
112+
"""Date of the snapshot."""
113+
return self._date
114+
115+
@property
116+
def impact_calc_data(self) -> dict:
117+
"""Convenience function for ImpactCalc class."""
118+
return {
119+
"exposures": self.exposure,
120+
"hazard": self.hazard,
121+
"impfset": self.impfset,
122+
}
123+
124+
@staticmethod
125+
def _convert_to_date(date_arg) -> datetime.date:
126+
"""Convert date argument of type int or str to a datetime.date object."""
127+
if isinstance(date_arg, int):
128+
# Assume the integer represents a year
129+
return datetime.date(date_arg, 1, 1)
130+
elif isinstance(date_arg, str):
131+
# Try to parse the string as a date
132+
try:
133+
return datetime.datetime.strptime(date_arg, "%Y-%m-%d").date()
134+
except ValueError:
135+
raise ValueError("String must be in the format 'YYYY-MM-DD'")
136+
elif isinstance(date_arg, datetime.date):
137+
# Already a date object
138+
return date_arg
139+
else:
140+
raise TypeError("date_arg must be an int, str, or datetime.date")
141+
142+
def apply_measure(self, measure: Measure) -> "Snapshot":
143+
"""Create a new snapshot by applying a Measure object.
144+
145+
This method creates a new `Snapshot` object by applying a measure on
146+
the current one.
147+
148+
Parameters
149+
----------
150+
measure : Measure
151+
The measure to be applied to the snapshot.
152+
153+
Returns
154+
-------
155+
The Snapshot with the measure applied.
156+
157+
"""
158+
159+
LOGGER.debug(f"Applying measure {measure.name} on snapshot {id(self)}")
160+
exp, impfset, haz = measure.apply(self.exposure, self.impfset, self.hazard)
161+
snap = Snapshot(exposure=exp, hazard=haz, impfset=impfset, date=self.date)
162+
snap._measure = measure
163+
return snap

‎climada/trajectories/test/test_snapshot.py‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
import datetime
2+
import unittest
3+
from unittest.mock import MagicMock
4+
5+
import numpy as np
6+
import pandas as pd
7+
8+
from climada.entity.exposures import Exposures
9+
from climada.entity.impact_funcs import ImpactFunc, ImpactFuncSet
10+
from climada.entity.measures.base import Measure
11+
from climada.hazard import Hazard
12+
from climada.trajectories.snapshot import Snapshot
13+
from climada.util.constants import EXP_DEMO_H5, HAZ_DEMO_H5
14+
15+
16+
class TestSnapshot(unittest.TestCase):
17+
18+
def setUp(self):
19+
# Create mock objects for testing
20+
self.mock_exposure = Exposures.from_hdf5(EXP_DEMO_H5)
21+
self.mock_hazard = Hazard.from_hdf5(HAZ_DEMO_H5)
22+
self.mock_impfset = ImpactFuncSet(
23+
[
24+
ImpactFunc(
25+
"TC",
26+
3,
27+
intensity=np.array([0, 20]),
28+
mdd=np.array([0, 0.5]),
29+
paa=np.array([0, 1]),
30+
)
31+
]
32+
)
33+
self.mock_measure = MagicMock(spec=Measure)
34+
self.mock_measure.name = "Test Measure"
35+
36+
# Setup mock return values for measure.apply
37+
self.mock_modified_exposure = MagicMock(spec=Exposures)
38+
self.mock_modified_hazard = MagicMock(spec=Hazard)
39+
self.mock_modified_impfset = MagicMock(spec=ImpactFuncSet)
40+
self.mock_measure.apply.return_value = (
41+
self.mock_modified_exposure,
42+
self.mock_modified_impfset,
43+
self.mock_modified_hazard,
44+
)
45+
46+
def test_init_with_int_date(self):
47+
snapshot = Snapshot(
48+
exposure=self.mock_exposure,
49+
hazard=self.mock_hazard,
50+
impfset=self.mock_impfset,
51+
date=2023,
52+
)
53+
self.assertEqual(snapshot.date, datetime.date(2023, 1, 1))
54+
55+
def test_init_with_str_date(self):
56+
snapshot = Snapshot(
57+
exposure=self.mock_exposure,
58+
hazard=self.mock_hazard,
59+
impfset=self.mock_impfset,
60+
date="2023-01-01",
61+
)
62+
self.assertEqual(snapshot.date, datetime.date(2023, 1, 1))
63+
64+
def test_init_with_date_object(self):
65+
date_obj = datetime.date(2023, 1, 1)
66+
snapshot = Snapshot(
67+
exposure=self.mock_exposure,
68+
hazard=self.mock_hazard,
69+
impfset=self.mock_impfset,
70+
date=date_obj,
71+
)
72+
self.assertEqual(snapshot.date, date_obj)
73+
74+
def test_init_with_invalid_date(self):
75+
with self.assertRaises(ValueError):
76+
Snapshot(
77+
exposure=self.mock_exposure,
78+
hazard=self.mock_hazard,
79+
impfset=self.mock_impfset,
80+
date="invalid-date",
81+
)
82+
83+
def test_init_with_invalid_type(self):
84+
with self.assertRaises(TypeError):
85+
Snapshot(
86+
exposure=self.mock_exposure,
87+
hazard=self.mock_hazard,
88+
impfset=self.mock_impfset,
89+
date=2023.5, # type: ignore
90+
)
91+
92+
def test_properties(self):
93+
snapshot = Snapshot(
94+
exposure=self.mock_exposure,
95+
hazard=self.mock_hazard,
96+
impfset=self.mock_impfset,
97+
date=2023,
98+
)
99+
100+
# We want a new reference
101+
self.assertIsNot(snapshot.exposure, self.mock_exposure)
102+
self.assertIsNot(snapshot.hazard, self.mock_hazard)
103+
self.assertIsNot(snapshot.impfset, self.mock_impfset)
104+
105+
# But we want equality
106+
pd.testing.assert_frame_equal(snapshot.exposure.gdf, self.mock_exposure.gdf)
107+
108+
self.assertEqual(snapshot.hazard.haz_type, self.mock_hazard.haz_type)
109+
self.assertEqual(snapshot.hazard.intensity.nnz, self.mock_hazard.intensity.nnz)
110+
self.assertEqual(snapshot.hazard.size, self.mock_hazard.size)
111+
112+
self.assertEqual(snapshot.impfset, self.mock_impfset)
113+
114+
def test_apply_measure(self):
115+
snapshot = Snapshot(
116+
exposure=self.mock_exposure,
117+
hazard=self.mock_hazard,
118+
impfset=self.mock_impfset,
119+
date=2023,
120+
)
121+
new_snapshot = snapshot.apply_measure(self.mock_measure)
122+
123+
self.assertIsNotNone(new_snapshot.measure)
124+
self.assertEqual(new_snapshot.measure.name, "Test Measure") # type: ignore
125+
self.assertEqual(new_snapshot.exposure, self.mock_modified_exposure)
126+
self.assertEqual(new_snapshot.hazard, self.mock_modified_hazard)
127+
self.assertEqual(new_snapshot.impfset, self.mock_modified_impfset)
128+
129+
130+
if __name__ == "__main__":
131+
TESTS = unittest.TestLoader().loadTestsFromTestCase(TestSnapshot)
132+
unittest.TextTestRunner(verbosity=2).run(TESTS)

0 commit comments

Comments
 (0)