Impact calc return impact forecast (#1179)

* Return impactForecast object in _return_impact
* Return impactForecast object in _return_empty
* Add full  impactcalc test for impactForecast
* Raise value error when computing impact with impact forecast without saving impact matrix
* Cosmetics: Improve error message, move test to own class
* Write nans for eai_exp and aai_agg when forecast is used

---------

Co-authored-by: luseverin <[email protected]>
Co-authored-by: Chahan Kropf <[email protected]>
Co-authored-by: Lukas Riedel <[email protected]>

Author: 3 people

climada/engine/impact_calc.py

@@ -29,6 +29,8 @@
 
 from climada import CONFIG
 from climada.engine.impact import Impact
+from climada.engine.impact_forecast import ImpactForecast
+from climada.hazard.forecast import HazardForecast
 
 LOGGER = logging.getLogger(__name__)
 
@@ -217,7 +219,7 @@ def _return_impact(self, imp_mat_gen, save_mat):
 
         Returns
         -------
-        Impact
+        Impact or ImpactForecast
             Impact Object initialize from the impact matrix
 
         See Also
@@ -230,12 +232,31 @@ def _return_impact(self, imp_mat_gen, save_mat):
             at_event, eai_exp, aai_agg = self.risk_metrics(
                 imp_mat, self.hazard.frequency
             )
+            if isinstance(self.hazard, HazardForecast):
+                eai_exp = np.full_like(eai_exp, np.nan, dtype=eai_exp.dtype)
+                aai_agg = np.full_like(aai_agg, np.nan, dtype=aai_agg.dtype)
+                LOGGER.warning(
+                    "eai_exp and aai_agg are undefined with forecasts. "
+                    "Setting them to NaN arrays."
+                )
+
         else:
+            if isinstance(self.hazard, HazardForecast):
+                raise ValueError(
+                    "Saving impact matrix is required when using HazardForecast."
+                    "Please set save_mat=True."
+                )
             imp_mat = None
             at_event, eai_exp, aai_agg = self.stitch_risk_metrics(imp_mat_gen)
-        return Impact.from_eih(
+
+        impact = Impact.from_eih(
             self.exposures, self.hazard, at_event, eai_exp, aai_agg, imp_mat
         )
+        if isinstance(self.hazard, HazardForecast):
+            return ImpactForecast.from_impact(
+                impact, self.hazard.lead_time, self.hazard.member
+            )
+        return impact
 
     def _return_empty(self, save_mat):
         """
@@ -248,21 +269,37 @@ def _return_empty(self, save_mat):
 
         Returns
         -------
-        Impact
+        Impact or ImpactForecast
             Empty impact object with correct array sizes.
         """
         at_event = np.zeros(self.n_events)
-        eai_exp = np.zeros(self.n_exp_pnt)
-        aai_agg = 0.0
+        if isinstance(self.hazard, HazardForecast):
+            eai_exp = np.full(self.n_exp_pnt, np.nan)
+            aai_agg = np.nan
+        else:
+            eai_exp = np.zeros(self.n_exp_pnt)
+            aai_agg = 0.0
+
         if save_mat:
             imp_mat = sparse.csr_matrix(
                 (self.n_events, self.n_exp_pnt), dtype=np.float64
             )
         else:
+            if isinstance(self.hazard, HazardForecast):
+                raise ValueError(
+                    "Saving impact matrix is required when using HazardForecast. "
+                    "Please set save_mat=True."
+                )
             imp_mat = None
-        return Impact.from_eih(
+
+        impact = Impact.from_eih(
             self.exposures, self.hazard, at_event, eai_exp, aai_agg, imp_mat
         )
+        if isinstance(self.hazard, HazardForecast):
+            return ImpactForecast.from_impact(
+                impact, self.hazard.lead_time, self.hazard.member
+            )
+        return impact
 
     def minimal_exp_gdf(
         self, impf_col, assign_centroids, ignore_cover, ignore_deductible

climada/engine/impact_forecast.py

@@ -90,6 +90,88 @@ def from_impact(
                 haz_type=impact.haz_type,
             )
 
+    @property
+    def at_event(self):
+        """Get the total impact for each member/lead_time combination."""
+        LOGGER.warning(
+            "at_event gives the total impact for one specific combination of member and "
+            "lead_time."
+        )
+        return self._at_event
+
+    @at_event.setter
+    def at_event(self, value):
+        """Set the total impact for each member/lead_time combination."""
+        self._at_event = value
+
+    def local_exceedance_impact(
+        self,
+        return_periods=(25, 50, 100, 250),
+        method="interpolate",
+        min_impact=0,
+        log_frequency=True,
+        log_impact=True,
+        bin_decimals=None,
+    ):
+        """Compution of local exceedance impact for given return periods is not
+        implemented for ImpactForecast.
+
+        See Also
+        --------
+        See :py:meth:`~climada.engine.impact.Impact.local_exceedance_impact`
+
+        Raises
+        ------
+        NotImplementedError
+        """
+
+        LOGGER.error("local_exceedance_impact is not defined for ImpactForecast")
+        raise NotImplementedError(
+            "local_exceedance_impact is not defined for ImpactForecast"
+        )
+
+    def local_return_period(
+        self,
+        threshold_impact=(1000.0, 10000.0),
+        method="interpolate",
+        min_impact=0,
+        log_frequency=True,
+        log_impact=True,
+        bin_decimals=None,
+    ):
+        """Compution of local return period for given impact thresholds is not
+        implemented for ImpactForecast.
+
+        See Also
+        --------
+        See :py:meth:`~climada.engine.impact.Impact.local_return_period`
+
+        Raises
+        -------
+        NotImplementedError
+        """
+
+        LOGGER.error("local_return_period is not defined for ImpactForecast")
+        raise NotImplementedError(
+            "local_return_period is not defined for ImpactForecast"
+        )
+
+    def calc_freq_curve(self, return_per=None):
+        """Computation of the impact exceedance frequency curve is not
+        implemented for ImpactForecast.
+
+        See Also
+        --------
+        See :py:meth:`~climada.engine.impact.Impact.calc_freq_curve`
+
+        Raises
+        ------
+        NotImplementedError
+        """
+
+        LOGGER.error("calc_freq_curve is not defined for ImpactForecast")
+        raise NotImplementedError("calc_freq_curve is not defined for ImpactForecast")
+
     def _check_sizes(self):
         """Check sizes of forecast data vs. impact data.