diff --git a/climada/engine/__init__.py b/climada/engine/__init__.py index ef8292f75d..1970f7706b 100755 --- a/climada/engine/__init__.py +++ b/climada/engine/__init__.py @@ -22,3 +22,4 @@ from .cost_benefit import * from .impact import * from .impact_calc import * +from .impact_forecast import ImpactForecast diff --git a/climada/engine/impact_forecast.py b/climada/engine/impact_forecast.py index eb69b01b5d..d534e32ead 100644 --- a/climada/engine/impact_forecast.py +++ b/climada/engine/impact_forecast.py @@ -18,3 +18,73 @@ Define Forecast variant of Impact. """ + +import logging + +import numpy as np + +from ..util import log_level +from ..util.forecast import Forecast +from .impact import Impact + +LOGGER = logging.getLogger(__name__) + + +class ImpactForecast(Forecast, Impact): + """An impact object with forecast information""" + + def __init__( + self, + *, + lead_time: np.ndarray | None, + member: np.ndarray | None, + **impact_kwargs, + ): + """Initialize the impact forecast. + + Parameters + ---------- + lead_time : np.ndarray, optional + The lead time associated with each event entry + member : np.ndarray, optional + The ensemble member associated with each event entry + impact_kwargs + Keyword-arguments passed to ~:py:class`climada.engine.impact.Impact`. + """ + # TODO: Maybe assert array lengths? + super().__init__(lead_time=lead_time, member=member, **impact_kwargs) + + @classmethod + def from_impact( + cls, impact: Impact, lead_time: np.ndarray | None, member: np.ndarray | None + ): + """Create an impact forecast from an impact object and forecast information. + + Parameters + ---------- + impact : climada.engine.impact.Impact + The impact object whose data to use in the forecast object + lead_time : np.ndarray, optional + The lead time associated with each event entry + member : np.ndarray, optional + The ensemble member associated with each event entry + """ + with log_level("WARNING", "climada.engine.impact"): + return cls( + lead_time=lead_time, + member=member, + event_id=impact.event_id, + event_name=impact.event_name, + date=impact.date, + frequency=impact.frequency, + frequency_unit=impact.frequency_unit, + coord_exp=impact.coord_exp, + crs=impact.crs, + eai_exp=impact.eai_exp, + at_event=impact.at_event, + tot_value=impact.tot_value, + aai_agg=impact.aai_agg, + unit=impact.unit, + imp_mat=impact.imp_mat, + haz_type=impact.haz_type, + ) diff --git a/climada/engine/test/test_impact.py b/climada/engine/test/test_impact.py index 38b3def3d8..f91fc2de98 100644 --- a/climada/engine/test/test_impact.py +++ b/climada/engine/test/test_impact.py @@ -47,26 +47,30 @@ STR_DT = h5py.special_dtype(vlen=str) -def dummy_impact(): - """Return an impact object for testing""" - return Impact( - event_id=np.arange(6) + 10, - event_name=[0, 1, "two", "three", 30, 31], - date=np.arange(6), - coord_exp=np.array([[1, 2], [1.5, 2.5]]), - crs=DEF_CRS, - eai_exp=np.array([7.2, 7.2]), - at_event=np.array([0, 2, 4, 6, 60, 62]), - frequency=np.array([1 / 6, 1 / 6, 1, 1, 1 / 30, 1 / 30]), - tot_value=7, - aai_agg=14.4, - unit="USD", - frequency_unit="1/month", - imp_mat=sparse.csr_matrix( +def impact_kwargs(): + return { + "event_id": np.arange(6) + 10, + "event_name": [0, 1, "two", "three", 30, 31], + "date": np.arange(6), + "coord_exp": np.array([[1, 2], [1.5, 2.5]]), + "crs": DEF_CRS, + "eai_exp": np.array([7.2, 7.2]), + "at_event": np.array([0, 2, 4, 6, 60, 62]), + "frequency": np.array([1 / 6, 1 / 6, 1, 1, 1 / 30, 1 / 30]), + "tot_value": 7, + "aai_agg": 14.4, + "unit": "USD", + "frequency_unit": "1/month", + "imp_mat": sparse.csr_matrix( np.array([[0, 0], [1, 1], [2, 2], [3, 3], [30, 30], [31, 31]]) ), - haz_type="TC", - ) + "haz_type": "TC", + } + + +def dummy_impact(): + """Return an impact object for testing""" + return Impact(**impact_kwargs()) def dummy_impact_yearly(): diff --git a/climada/engine/test/test_impact_forecast.py b/climada/engine/test/test_impact_forecast.py index 8fe6fa72d4..20f5420ae3 100644 --- a/climada/engine/test/test_impact_forecast.py +++ b/climada/engine/test/test_impact_forecast.py @@ -18,3 +18,57 @@ Tests for Impact Forecast. """ + +import numpy as np +import numpy.testing as npt +import pandas as pd +import pytest +from scipy.sparse import csr_matrix + +from climada.engine import Impact, ImpactForecast + +from .test_impact import impact_kwargs as imp_kwargs + + +@pytest.fixture +def impact_kwargs(): + return imp_kwargs() + + +@pytest.fixture +def impact(impact_kwargs): + return Impact(**impact_kwargs) + + +def assert_impact_kwargs(impact: Impact, **kwargs): + for key, value in kwargs.items(): + attr = getattr(impact, key) + if isinstance(value, (np.ndarray, list)): + npt.assert_array_equal(attr, value) + elif isinstance(value, csr_matrix): + npt.assert_array_equal(attr.todense(), value.todense()) + else: + assert attr == value + + +class TestImpactForecastInit: + lead_time = pd.date_range("2000-01-01", "2000-01-02", periods=6).to_numpy() + member = np.arange(6) + + def test_impact_forecast_init(self, impact_kwargs): + forecast1 = ImpactForecast( + lead_time=self.lead_time, + member=self.member, + **impact_kwargs, + ) + npt.assert_array_equal(forecast1.lead_time, self.lead_time) + npt.assert_array_equal(forecast1.member, self.member) + assert_impact_kwargs(forecast1, **impact_kwargs) + + def test_impact_forecast_from_impact(self, impact, impact_kwargs): + forecast = ImpactForecast.from_impact( + impact, lead_time=self.lead_time, member=self.member + ) + npt.assert_array_equal(forecast.lead_time, self.lead_time) + npt.assert_array_equal(forecast.member, self.member) + assert_impact_kwargs(forecast, **impact_kwargs)