Add mean max min methods to impact forecast and tests

Author: luseverin

climada/engine/impact_forecast.py

@@ -22,6 +22,7 @@
 import logging
 
 import numpy as np
+import scipy.sparse as sparse
 
 from ..util import log_level
 from ..util.forecast import Forecast
@@ -152,3 +153,137 @@ def calc_freq_curve(self, return_per=None):
 
         LOGGER.error("calc_freq_curve is not defined for ImpactForecast")
         raise NotImplementedError("calc_freq_curve is not defined for ImpactForecast")
+
+    def _reduce_attrs(self, reduce_method: str):
+        """
+        Reduce the attributes of an ImpactForecast to a single value.
+
+        Attributes are modified as follows:
+        - event_id: set to [0]
+        - event_name: set to [reduce_method]
+        - date: set to the minimum value
+        - frequency: set to 0
+
+        Parameters
+        ----------
+        reduce_method : str
+            The reduction method used to reduce the attributes.
+        """
+        red_event_id = np.asarray([0])
+        red_event_name = np.asarray([reduce_method])
+        red_date = np.array([self.date.min()])
+        red_frequency = np.array([0])
+        return red_event_id, red_event_name, red_date, red_frequency
+
+    def min(self):
+        """
+        Reduce the impact matrix and at_event of an ImpactForecast to the minimum
+        value.
+
+        Parameters
+        ----------
+        None
+
+        Returns
+        -------
+        None
+        """
+        red_imp_mat = sparse.csr_matrix(self.imp_mat.min(axis=0))
+        red_at_event = np.array([red_imp_mat.sum()])
+        red_event_id, red_event_name, red_date, red_frequency = self._reduce_attrs(
+            "min"
+        )
+        return ImpactForecast(
+            lead_time=self.lead_time,
+            member=self.member,
+            event_id=red_event_id,
+            event_name=red_event_name,
+            date=red_date,
+            frequency=red_frequency,
+            frequency_unit=self.frequency_unit,
+            coord_exp=self.coord_exp,
+            crs=self.crs,
+            eai_exp=self.eai_exp,
+            at_event=red_at_event,
+            tot_value=self.tot_value,
+            aai_agg=self.aai_agg,
+            unit=self.unit,
+            imp_mat=red_imp_mat,
+            haz_type=self.haz_type,
+        )
+
+    def max(self):
+        """
+        Reduce the impact matrix and at_event of an ImpactForecast to the maximum
+        value.
+
+        Parameters
+        ----------
+        None
+
+        Returns
+        -------
+        None
+        """
+        red_imp_mat = sparse.csr_matrix(self.imp_mat.max(axis=0))
+        red_at_event = np.array([red_imp_mat.sum()])
+        red_event_id, red_event_name, red_date, red_frequency = self._reduce_attrs(
+            "max"
+        )
+        return ImpactForecast(
+            lead_time=self.lead_time,
+            member=self.member,
+            event_id=red_event_id,
+            event_name=red_event_name,
+            date=red_date,
+            frequency=red_frequency,
+            frequency_unit=self.frequency_unit,
+            coord_exp=self.coord_exp,
+            crs=self.crs,
+            eai_exp=self.eai_exp,
+            at_event=red_at_event,
+            tot_value=self.tot_value,
+            aai_agg=self.aai_agg,
+            unit=self.unit,
+            imp_mat=red_imp_mat,
+            haz_type=self.haz_type,
+        )
+
+    def mean(self):
+        """
+        Reduce the impact matrix and at_event of an ImpactForecast to the mean value.
+
+        The mean value is computed by taking the mean of the impact matrix along the
+        exposure points axis (axis=1) and then taking the mean of the resulting array.
+
+        Parameters
+        ----------
+        None
+
+        Returns
+        -------
+        None
+        """
+        red_imp_mat = sparse.csr_matrix(self.imp_mat.mean(axis=0))
+        red_at_event = np.array([red_imp_mat.sum()])
+        red_event_id, red_event_name, red_date, red_frequency = self._reduce_attrs(
+            "mean"
+        )
+        return ImpactForecast(
+            lead_time=self.lead_time,
+            member=self.member,
+            event_id=red_event_id,
+            event_name=red_event_name,
+            date=red_date,
+            frequency=red_frequency,
+            frequency_unit=self.frequency_unit,
+            coord_exp=self.coord_exp,
+            crs=self.crs,
+            eai_exp=self.eai_exp,
+            at_event=red_at_event,
+            tot_value=self.tot_value,
+            aai_agg=self.aai_agg,
+            unit=self.unit,
+            imp_mat=red_imp_mat,
+            haz_type=self.haz_type,
+        )

climada/engine/test/test_impact_forecast.py

@@ -115,3 +115,39 @@ def test_impact_forecast_blocked_methods(impact_forecast):
 
     with pytest.raises(NotImplementedError):
         impact_forecast.calc_freq_curve(np.array([10, 50, 100]))
+
+
+def test_impact_forecast_mean_min_max(impact_forecast):
+    """Check mean, min, and max methods for ImpactForecast"""
+    imp_fcst_mean = impact_forecast.mean()
+    imp_fcst_min = impact_forecast.min()
+    imp_fcst_max = impact_forecast.max()
+    # sparse.csr_matrix(
+    #        np.array([[0, 0], [1, 1], [2, 2], [3, 3], [30, 30], [31, 31]])
+
+    # assert imp_mat
+    npt.assert_array_equal(
+        imp_fcst_mean.imp_mat.todense(), impact_forecast.imp_mat.todense().mean(axis=0)
+    )
+    npt.assert_array_equal(imp_fcst_min.imp_mat.todense(), np.array([[0, 0]]))
+    npt.assert_array_equal(imp_fcst_max.imp_mat.todense(), np.array([[31, 31]]))
+    # assert at_event
+    npt.assert_array_equal(
+        imp_fcst_mean.at_event, impact_forecast.at_event.mean()
+    )  # 134/6
+    npt.assert_array_equal(imp_fcst_min.at_event, impact_forecast.at_event.min())
+    npt.assert_array_equal(imp_fcst_max.at_event, impact_forecast.at_event.max())
+
+    # check that attributes where reduced correctly
+    assert imp_fcst_mean.event_name[0] == "mean"
+    assert imp_fcst_min.event_name[0] == "min"
+    assert imp_fcst_max.event_name[0] == "max"
+    assert imp_fcst_mean.event_id[0] == 0
+    assert imp_fcst_min.event_id[0] == 0
+    assert imp_fcst_max.event_id[0] == 0
+    assert imp_fcst_mean.frequency == 0
+    assert imp_fcst_min.frequency == 0
+    assert imp_fcst_max.frequency == 0
+    assert imp_fcst_mean.date == impact_forecast.date.min()
+    assert imp_fcst_min.date == impact_forecast.date.min()
+    assert imp_fcst_max.date == impact_forecast.date.min()