Impact forecast class (#1168)

* Add ImpactForecast
* Add unit tests

Author: peanutfun

climada/engine/__init__.py

@@ -22,3 +22,4 @@
 from .cost_benefit import *
 from .impact import *
 from .impact_calc import *
+from .impact_forecast import ImpactForecast

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,
+            )

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():

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)